rich progress and multiprocessing
Tags: rich, python, progress, multiprocessing
Published: 2022-03-24 16:39:13

---
title: rich progress and multiprocessing
published: true
date: 2022-03-24 00:00:00 UTC
tags: rich,python,progress,multiprocessing
canonical_url: https://www.deanmontgomery.com/2022/03/24/rich-progress-and-multiprocessing/
---

How to use [rich](https://github.com/Textualize/rich#rich-library) with python’s [multiprocessing](https://docs.python.org/3/library/multiprocessing.html):

- What is this? 
  - Track progress of long running tasks when using `multiprocessing`
- Why would you want to do this? 
  - When you are doing lots of things with `multiprocessing` and each task can take a long time - feedback makes it easier to see that things are actually happening


How to do it:

```
import multiprocessing
import random
from concurrent.futures import ProcessPoolExecutor
from time import sleep

from rich import progress

def long_running_fn(progress, task_id):
    len_of_task = random.randint(3, 20) # take some random length of time
    for n in range(0, len_of_task):
        sleep(1) # sleep for a bit to simulate work
        progress[task_id] = {"progress": n + 1, "total": len_of_task}

if __name__ == " __main__":
    n_workers = 8 # set this to the number of cores you have on your machine

    with progress.Progress(
        "[progress.description]{task.description}",
        progress.BarColumn(),
        "[progress.percentage]{task.percentage:>3.0f}%",
        progress.TimeRemainingColumn(),
        progress.TimeElapsedColumn(),
        refresh_per_second=1, # bit slower updates
    ) as progress:
        futures = [] # keep track of the jobs
        with multiprocessing.Manager() as manager:
            # this is the key - we share some state between our 
            # main process and our worker functions
            _progress = manager.dict()
            overall_progress_task = progress.add_task("[green]All jobs progress:")

            with ProcessPoolExecutor(max_workers=n_workers) as executor:
                for n in range(0, 20): # iterate over the jobs we need to run
                    # set visible false so we don't have a lot of bars all at once:
                    task_id = progress.add_task(f"task {n}", visible=False)
                    futures.append(executor.submit(long_running_fn, _progress, task_id))

                # monitor the progress:
                while (n_finished := sum([future.done() for future in futures])) < len(
                    futures
                ):
                    progress.update(
                        overall_progress_task, completed=n_finished, total=len(futures)
                    )
                    for task_id, update_data in _progress.items():
                        latest = update_data["progress"]
                        total = update_data["total"]
                        # update the progress bar for this task:
                        progress.update(
                            task_id,
                            completed=latest,
                            total=total,
                            visible=latest < total,
                        )

                # raise any errors:
                for future in futures:
                    future.result()

```