multiprocessing with map_async and progress bar

Question

Is it possible to have progress bar with map_async from multiprocessing:

toy example:

from multiprocessing import Pool
import tqdm

def f(x):
    print(x)
    return x*x

n_job = 4
with Pool(processes=n_job) as pool:
    results = pool.map_async(f, range(10)).get()
print(results)

something like this:

data = []

with Pool(processes=10) as pool:
        for d in tqdm.tqdm(
                pool.imap(f, range(10)),
                total=10):
            data.append(d)

Answer 1

There are a couple of ways of achieving what you want that I can think of:

1. Use apply_async with a callback argument to update the progress bar as each result becomes available.
2. Use imap and as you iterate the results you can update the progress bar.

There is a slight problem with imap in that the results must be returned in task-submission order, which is of course what you want. But that order does not necessarily reflect the order in which the submitted tasks complete so the progress bar is not necessarily getting updated as frequently as it otherwise might. But I will show that solution first since it is the simplest and probably adequate:

from multiprocessing import Pool
import tqdm

def f(x):
    import time
    time.sleep(1) # for demo purposes
    return x*x

# Required by Windows:
if __name__ == '__main__':
    pool_size = 4
    results = []
    with Pool(processes=pool_size) as pool:
        with tqdm.tqdm(total=10) as pbar:
            for result in pool.imap(f, range(10)):
                results.append(result)
                pbar.update()
    print(results)

The solution that uses apply_async:

from multiprocessing import Pool
import tqdm

def f(x):
    import time
    time.sleep(1) # for demo purposes
    return x*x

# Required by Windows:
if __name__ == '__main__':
    def my_callback(_):
        # We don't care about the actual result.
        # Just update the progress bar:
        pbar.update()

    pool_size = 4
    with Pool(processes=pool_size) as pool:
        with tqdm.tqdm(total=10) as pbar:
            async_results = [pool.apply_async(f, args=(x,), callback=my_callback) for x in range(10)]
            results = [async_result.get() for async_result in async_results]
    print(results)

Answer 2

I think this is it:

from multiprocessing import Pool
import tqdm

def f(x):
    return x*x

n_job = 4

data = []
with Pool(processes=10) as pool:
        for d in tqdm.tqdm(
                pool.map_async(f, range(10)).get(),
                total=10):
            data.append(d)
print(data)