Python tqdm process_map: Append list shared between processes?

Question

I want to share a list to append output from parallel threads, started by process_map from tqdm. (The reason why I want to use process_map is the nice progress indicator and the max_workers= option.)

I have tried to use from multiprocessing import Manager to create the shared list, but I am doing something wrong here: My code prints an empty shared_list, but it should print a list with 20 numbers, correct order is not important.

Any help would be greatly appreciated, thank you in advance!

import time
from tqdm.contrib.concurrent import process_map
from multiprocessing import Manager


shared_list = []

def worker(i):
    global shared_list
    time.sleep(1)
    shared_list.append(i)

if __name__ == '__main__':
    manager = Manager()
    shared_list = manager.list()

    process_map(worker, range(20), max_workers=5)
    print(shared_list)

Answer 1

You didn't specify what platform you are running under (you are supposed to tag your question with your platform whenever you tag a question with multiprocessing), but it appears you are running under a platform that uses spawn to create new processes (such as Windows). This means that when a new process is launched, an empty address space is created, a new Python interpreter is launched and the source is re-executed from the top.

So although you have in the block that begins if __name__ == '__main__': assigned to shared_list a managed list, each process in the pool that is created will be executing shared_list = [] clobbering your initial assignment.

You can pass shared_list as the first argument to your worker function:

import time
from tqdm.contrib.concurrent import process_map
from multiprocessing import Manager
from functools import partial

def worker(shared_list, i):
    time.sleep(1)
    shared_list.append(i)

if __name__ == '__main__':
    manager = Manager()
    shared_list = manager.list()

    process_map(partial(worker, shared_list), range(20), max_workers=5)
    print(shared_list)

If process_map supported the initializer and initargs arguments in the same way as the ProcessPoolExecutor class does (it appears that it does not), then you could do:

import time
from tqdm.contrib.concurrent import process_map
from multiprocessing import Manager

def init_pool(the_list):
    global shared_list
    shared_list = the_list

def worker(i):
    time.sleep(1)
    shared_list.append(i)

if __name__ == '__main__':
    manager = Manager()
    shared_list = manager.list()

    process_map(worker, range(20), max_workers=5, initializer=init_pool, initargs=(shared_list,))
    print(shared_list)

Comment

This has nothing to do per se with your original problem, but for this type of problem you might want to consider using instead of a managed list to which your worker function (coincidentally named worker) appends elements and the order in which the elements are appended is non-deterministic since you do not have any control over the scheduling of the pool processes, a multiprocessing.Array instance initialized as follows:

shared_list = Array('i', [0] * 20, lock=False)

And then you worker function becomes:

def worker(i):
    time.sleep(1)
    shared_list[i] = i

Here the array is being stored in shared memory and there is not even a need for locked access since each invocation of worker is accessing a different index of the array. Accessing elements of a shared memory Array is much faster than accessing elements of a managed list. The only problem is that references to shared memory cannot be passed as arguments and we saw that process_map does not support the initializer and initargs arguments. So you would have to use lower-level methods. For example:

import time
from multiprocessing import Pool, Array
from tqdm import tqdm

def init_pool(the_list):
    global shared_list
    shared_list = the_list

def worker(i):
    time.sleep(1)
    shared_list[i] = i

if __name__ == '__main__':
    # Preallocate 20 slots for the array in shared memory
    # And we don't require a lock if each worker invocation is accessing a different Array index:
    args = range(20)
    shared_list = Array('i', [0] * len(args), lock=False)

    with tqdm(total=len(args)) as pbar:
        pool = Pool(5, initializer=init_pool, initargs=(shared_list,))
        for result in pool.imap_unordered(worker, args):
            pbar.update(1)
    # print out elements one at a time:
    for elem in shared_list:
        print(elem)
    # print out all elements at once (must first convert to a regular list):
    print(list(shared_list))

Comment 2

I would avoid using process_map. This function is based on the map method of the ProcessPoolExecutor.map method, which is required to return results in an order corresponding to the elements of the iterable being passed, not in the order of completion. Imagine what happens if for some reason the process that is processing the first task submitted, i value 0 in our case, takes a very long time to process and turns out to be the last task completed. You would see the tqdm progress bar do nothing for a long time until that first submitted task completed. But when that happens we know that all the other submitted tasks have already completed and so the progress bar would jump from 0 to 100% instantaneously. Modify function worker as follows to see this in action:

def worker(shared_list, i):
    if i == 0:
        time.sleep(5)
    else:
        time.sleep(.25)
    shared_list.append(i)

The code version I offered above that uses Pool.imap_unordered is allowed to returns results out of order and with the default chunksize value of 1, it will be in the order of completion. The progress bar will more smoothly progress.

Comment 3

There also seems to be a bug in tqdm. The following program demonstrates how you would use the low-level tqdm calls this time with the concurrent.futures module. Unfortunately, its ProcessPoolExecutor class (for multiprocessing) and ThreadPoolExecutor class (for multithreading) does not have an equivalent to the imap_unordered method. You have to use the submit method (whose multiprocessing.pool.Pool analog would be the apply_async method), which returns a Future instance on which you can call the result method to block for completion and to return the result of the submitted task). You would submit a bunch of tasks and store the returned Future instances perhaps in a list and then use the as_completed function call to be returned from that list the next completed Future instance that has completed.

This demo uses threading and creates a thread pool who size is 20 and submits 20 tasks, so all tasks should start at the same time. The sleep time for worker1 is set to vary so that the smaller the value for the i argument, the longer the sleep time is. This program creates the pool and submits the tasks 4 times. The first time, the return values are just printed. The second time a tqdm progress bar is used. The results are as you would expect. The third time worker2 is used with the tqdm progress bar. The difference is that for all values of i != 0 the sleep time is a constant (.25 seconds), so that for i values 1, 2, ... 19, the tasks should complete at roughly the same time. You would therefore expect to see the progress bar within a very short time jump to 95% and then wait for the i == 0 task to complete. What you observe, however, is the opposite. The progress bar goes to 5% and hangs there for a long time and then jumps to 100%. The fourth case is using worker2 with my own "progress bar", which behaves as you would expect.

This is tqdm 4.61.1 under Python 3.8.5. I have tested this under Windows and Linux. Does anyone have an explanation for this behavior?

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm
import sys

class MyProgressBar:
    def __init__(self, n_tasks):
        self._task_count = n_tasks
        self._completed = 0
        self.update()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(file=sys.stderr)
        return False

    def update(self, count=0):
        self._completed += count
        print(f'\r{self._completed} of {self._task_count} task(s) completed.', end='', flush=True)

def worker1(i):
    if i == 0:
        time.sleep(8)
    else:
        time.sleep(5 - i/5)
    return i

def worker2(i):
    if i == 0:
        time.sleep(8)
    else:
        time.sleep(.25)
    return i

if __name__ == '__main__':
    args = range(20)

    with ThreadPoolExecutor(max_workers=20) as pool:
        futures = [pool.submit(worker1, arg) for arg in args]
        for future in as_completed(futures):
            print(future.result())

    with ThreadPoolExecutor(max_workers=20) as pool:
        with tqdm(total=len(args)) as pbar:
            futures = [pool.submit(worker1, arg) for arg in args]
            for future in as_completed(futures):
                future.result()
                pbar.update(1)

    with ThreadPoolExecutor(max_workers=20) as pool:
        with tqdm(total=len(args)) as pbar:
            futures = [pool.submit(worker2, arg) for arg in args]
            for future in as_completed(futures):
                future.result()
                pbar.update(1)

    # Works with my progress "bar":
    with ThreadPoolExecutor(max_workers=20) as pool:
        with MyProgressBar(len(args)) as pbar:
            futures = [pool.submit(worker2, arg) for arg in args]
            for future in as_completed(futures):
                future.result()
                pbar.update(1)