Adding jobs to pool dynamically or whenever a core is freed up – Ubuntu

Please see my comment posted on your question. It’s not entirely clear what your "listener" is attempting to do because you have posted incomplete code. It’s also not clear, for the same reason, how arguments for any given job submission is generated. But I will attempt to give you some ideas that may be useful to you.

Let’s supposed that you had a generator function that can generate arguments for your , which we will arbitrarily name calculate:

def generate_arguments():
    while True:
        # Generate next set of arguments:
        ... # Code omitted
        yield arg1, arg2, arg3

Now we can use a special subclass of multiprocessing.pool.Pool I called BoundedQueuePool for which you can repeatedly submit jobs with apply_async but will throttle job submission so that the number of total jobs sitting on the pool’s input job queue does not by default exceed twice the number of pool processes. We continue to generate arguments and submit new jobs until a callback function that gets invoked whenever a new result has been generated. This function will detect whether the results have converged or not and (1) Set the global flag converged to True so that no more jobs are submitted and (2) A call to terminate on the pool to kill any submitted jobs that have not yet completed. This happens implicitly when the with BoundedQueueProcessPool(8) as pool: exits.

Putting it all together:

import multiprocessing.pool
import multiprocessing

class BoundedQueuePool:
    def __init__(self, limit, semaphore_type):
        self._semaphore = semaphore_type(limit)

    def release(self, result, callback=None):
        self._semaphore.release()
        if callback:
            callback(result)

    def apply_async(self, func, args=(), kwds={}, callback=None, error_callback=None):
        self._semaphore.acquire()
        callback_fn = self.release if callback is None else lambda result: self.release(result, callback=callback)
        error_callback_fn = self.release if error_callback is None else lambda result: self.release(result, callback=callback)
        return super().apply_async(func, args, kwds, callback=callback_fn, error_callback=error_callback_fn)

class BoundedQueueProcessPool(BoundedQueuePool, multiprocessing.pool.Pool):
    def __init__(self, *args, max_waiting_tasks=None, **kwargs):
        multiprocessing.pool.Pool.__init__(self, *args, **kwargs)
        if max_waiting_tasks is None:
            max_waiting_tasks = self._processes
        elif max_waiting_tasks < 0:
            raise ValueError(f'Invalid negative max_waiting_tasks value: {max_waiting_tasks}')
        limit = self._processes + max_waiting_tasks
        BoundedQueuePool.__init__(self, limit, multiprocessing.BoundedSemaphore)


if __name__ == '__main__': # Required for Windows
    def generate_arguments():
        while True:
            # Generate next set of arguments:
            ... # Code omitted
            yield arg1, arg2, arg3

    converged = False

    def my_callback(result):
        global converged
        ... # Code omitted
        # We have determined that the results have converged:
        converged = True

    with BoundedQueueProcessPool(8) as pool:
        for arg1, arg2, arg3 in generate_arguments():
            if converged:
                break
            pool.apply_async(calculate, args=(arg1, arg2, arg3), callback=my_callback)

Explanation

Suppose you have a large number of tasks to submit to a processing pool via method multiprocessing.pool.Pool.apply_async and that you can generate these tasks much faster than pool can process each task and generate a result. You would have the situation where the input task queue of the pool, which holds tasks waiting to be processed, will continue to grow and consume memory. This is not an issue if the number of tasks that will eventually be put on queue is not that great and the amount of data (the arguments to the worker function(s)) is not so large and you have adequate memory. But take the example of where you have a multi-gigabyte text file where each line of that file represents another task to be submitted. If the main process goes into a loop reading the file line by line submitting each one to the pool, you will soon exhaust all of memory. Or take the case where you have potentially an unlimited number of tasks you are submitting. In your example you might be in a loop generating tasks for as long as it takes until a certain result is produced.

In these cases you would like a method to "throttle" the submission of tasks. Let’s assume the pool as N processes. Then there is really no need for the input queue to have more N * 2 tasks on the input queue at any point in time. I specify N * 2 instead of N just to ensure that when a pool process becomes idle there is a task on the queue ready to be fetched and executed rather than having to wait for the main process to generate another task. But whether it is N or N * 2 tasks or a larger number really isn’t anything to worry about for the purposes of this discussion.

So class BoundedQueueProcessPool, a specialization of multiprocessing.pool.Pool, supports a modified version of method apply_async. "Under the covers" it creates a semaphore initialized to size N * 2 (by default) where N is the pool size. When you call apply_async the code first does an acquire method call on the semaphore. You can do this N * 2 times before the semaphore count goes to 0 and a subsequent call will block. However, whenever a task completes, thus making a pool process idle and ready to take a new task from the input queue, a call to release is made on the semaphore thus allowing a blocked apply_async call to acquire the semaphore and complete the submission of a new task.

To be able to support both "bounded queue-size" multiprocessing pools and bounded queue-size multithreading pools, much of the aforementioned logic resides in a parent class BoundedQueuePool, which is the actual direct child of Pool and BoundedQueueProcessPool is then a child of BoundedQueuePool. This make it easy to create another class BoundedQueueThreadPool (not shown) that is a child of BoundedQueuePool to handle bounded queue-size multithreading pools.

The important point of the code is that the main process is in a loop calling apply_async with an inexhaustible collection of tasks it will submit until a certain result is detected. So as to not allow the queue size to grow unwieldly large, we throttle task submission so that the queue size never exceeds a certain amount.

It appears that the solution posted by Ahmed AEK is also trying to keep the pool size limited in a different way by first submitting a fixed number of tasks and then only waiting until a task completes submitting another task. The difference with my solution is that the logic is encapsulated in a way that makes it (in my opinion) more readily reusable in other solutions you might have. All you have to do is change the class you are using for the pool and no other code changes are required.