Python 进度条与 fork 进程结合使用

Question

我正在尝试将进度条合并到使用 os.fork 并行操作的一段代码中。我尝试过使用丰富的进度条，但由于跨进程处理进度更新的方式，它不起作用，即子进程内的进度条更新未反映在父进程中。

我的代码看起来像这样：

from rich.progress import Progress
import os
with Progress() as progress:
    task = progress.add_task(total=len(args_list))
    for balanced_batch in even_batches_it:
        if os.fork():
            # do something
        else:
            try:
                for my_tuple in balanced_batch:
                    do_something(my_tuple)
                progress.advance(task, advance=len(balanced_genes_batch))
            except Exception as exception:
              # do something

非常感谢任何反馈/建议！

Answer 1

根据我的经验，最好只创建一个专门用于执行此任务的 python 文件。每当我想在代码中添加进度条时，它都会让事情变得容易得多，因为

threading

并不真正起作用。

我创建了一些代码，以便当您向其中输入函数时，它会在进度条旁边运行该函数：

from rich.progress import Progress
import time
import inspect

def run(func, sleep=0):
    update = lambda progress, task: progress.update(task, advance=1)

    function = inspect.getsource(func)
    mylist = function.split('\n')
    mylist.remove(mylist[0])
    length = len(mylist)
    for num, line in enumerate(mylist):
        mylist[num] = line[8:] + "\nupdate(progress, task)\ntime.sleep(sleep)" #update the progress bar

    myexec = ''
    for line in mylist:
        myexec += line + '\n'

    with Progress() as progress:
        task = progress.add_task("Working...", total=length)
        exec(myexec)

if __name__ == '__main__':
    def myfunc():
        #chose to import these modules for progress bar example since they take a LONG time to import on my computer
        import pandas
        import openai

    run(myfunc)

这段代码效率有点低，因为它使用了

exec()

。不过，它的威力却非常大！通过在要运行的函数末尾的每一行添加

update(progress, task)

，进度条会更新函数的每一行。以下是实现代码的方法：

将此示例代码保存为文件，命名为任意名称（例如

progress_bar.py

）。现在从 python 文件导入函数

run()

。现在你可以运行你的代码了：

from progress_bar import run
import os

def myfunc():
    for balanced_batch in even_batches_it:
        if os.fork():
            # do something
        else:
            try:
                for my_tuple in balanced_batch:
                    do_something(my_tuple)
                progress.advance(task, advance=len(balanced_genes_batch))
            except Exception as exception:
              # do something

run(myfunc) #runs this function so the progress bar will update alongside it, updating every line of code in the function

最后要注意的是，您的代码实际上是为循环构建的，因此您不需要我的函数来运行它（根据文档），但考虑到您的当前进度条无法正常工作。