multiprocessing_resizeable_pool.py

from multiprocessing.pool import Pool, worker



class DynamicPool(Pool):

    def __init__(self, processes=None, initializer=None, initargs=()):
        super(DynamicPool, self).__init__(processes=processes,
                                          initializer=initializer,
                                          initargs=initargs)
        self._initializer = initializer
        self._initargs = initargs

    def _add_worker(self):
        w = self.Process(target=worker,
                         args=(self._inqueue, self._outqueue,
                               self._initializer, self._initargs))
        self._pool.append(w)
        w.name = w.name.replace("Process", "PoolWorker")
        w.daemon = True
        w.start()

    def increment_size(self, n=1):
        [self._add_worker() for i in xrange(n)]


def get_logger():
    from multiprocessing import log_to_stderr
    import logging
    logger = log_to_stderr()
    logger.setLevel(logging.INFO)
    return logger


def myfunc(i):
    get_logger().info("i * i = %d" % (i*i))
    return i * i


if __name__ == "__main__":
    p = DynamicPool(2)
    res = p.map(myfunc, [2, 4, 6, 8, 10])

    p.increment_size(5)
    res = p.map(myfunc, [2, 4, 6, 8, 10])