How to use multiprocessing.Pool async without callbacks?

Question

So, I have a problem which I assume must be common:

I'd like to parallelize a script with a multiprocessing.Pool, handing inputs to the pool, having it process them in parallel, and receive the outputs in the parent process.

apply_async() looks like the best fit for what I want to do. But I can't just give a callback function, since in the end I want to print all the results to a single file. I think handing it a callback which prints to a single filehandle will result in jumbled results (not even sure I can pass a filehandle between processes like that).

So how's the best way to submit inputs to the Pool, then receive the outputs and handle them in the main process? At the moment I'm just collecting the AsyncResult objects in a list and periodically iterating through it, calling a .get() method on each.

Update

I'll clarify a few parameters of my problem in response to comments:

1. @martineau and @Juggernaut: By not "jumbled" I mean I'd really like to preserve the order of the input so that the output is in the same order.

2. @RolandSmith and @martineau: My main process is just reading inputs from a file, handing them to the pool, and printing the results. I could just call .apply(), but then the main process is waiting for the function to complete before it proceeds. I'm using multiprocessing to reap the benefits of parallelization and have many inputs processed simultaneously.

Answer 1

To kind of answer your question, I don't think you can do what you want without a callback.

You want the results computed asynchronously, yet printed in the same order as the inputs. This implies not only having to wait until all of the inputs are processed before printing them, but also some way to know their relative position in the inputs so they can be sorted back into that order before outputting them.

So here's how to do it with one. As I previous said, the slightly tricky part is that the results must include something that indicates the corresponding position of the input so the results can be sorted in a similar order before printing. Because of that requirement, the script must wait until all the inputs have been processed.

Note that despite this, you are getting the benefits of parallel processing in the sense that the individual results themselves are being created by concurrent processes.

import multiprocessing as mp
from random import randint
from time import sleep

def my_func(*args):
    print('my_func:', args)
    index, x = args
    sleep(randint(1, 3))  # Take a varying amount of time to finish.
    return index, x*x  # Return result index and value.

if __name__ == '__main__':

    result_list = []

    def errorhandler(exc):
        print('Exception:', exc)

    def log_result(result):
        # This is called whenever my_func() returns a result.
        # result_list is modified only by the main process, not the pool workers.
        result_list.append(result)

    pool = mp.Pool()
    for i in range(10):
        pool.apply_async(my_func, args=(i, i*2), callback=log_result,
                         error_callback=errorhandler)
    pool.close()
    pool.join()  # Wait for all subprocesses to finish.

    print('result_list:', result_list)
    sorted_results = [x[1] for x in sorted(result_list)]
    print('sorted results:', sorted_results)

Output:

my_func: (5, 10)
my_func: (1, 2)
my_func: (4, 8)
my_func: (7, 14)
my_func: (3, 6)
my_func: (9, 18)
my_func: (0, 0)
my_func: (6, 12)
my_func: (2, 4)
my_func: (8, 16)
result_list: [(2, 16), (3, 36), (5, 100), (1, 4), (4, 64), (7, 196), (9, 324), (0, 0), (6, 144), (8, 256)]
sorted results: [0, 4, 16, 36, 64, 100, 144, 196, 256, 324]

Answer 2

As you requested in a comment, here's code showing how to do it with Pool.map() instead of Pool.async()—which seems like a much better fit for this problem given the need to wait for all the results before further output processing can be done (e.g. it needs to be in the same order as the input).

As you can see it requires substantially less code and doesn't require sorting the results before outputting them (so is probably faster, too).

import multiprocessing as mp
from random import randint
from time import sleep

def my_func(x):
    print('my_func:', x)
    sleep(randint(1, 3))  # Take a varying amount of time to finish.
    return x*x

if __name__ == '__main__':

    input_data = range(10)
    with mp.Pool(10) as pool:
        result_list = pool.map(my_func, input_data)  # Blocks until finished.

    print('result_list:', result_list)  # Will be in same order as input_data.

Output:

my_func: 3
my_func: 2
my_func: 1
my_func: 0
my_func: 8
my_func: 5
my_func: 7
my_func: 6
my_func: 4
my_func: 9
result_list: [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

Answer 3

I think handing it a callback which prints to a single filehandle will result in jumbled results

The solution is to populate a Queue in your callback with the result and then fetch them later. Queues are thread-safe so you don't have to worry about the jumbled results you're talking about.

from queue import Queue
results = Queue()

def callback(result):
    results.put(result)

later

item = results.get()