parallel.futures：如何在2个单独的线程中提交任务和处理Future？

如何解决parallel.futures：如何在2个单独的线程中提交任务和处理Future？

我正在尝试改善小型CLI程序的交互式输出，使其在目录中处理文件，并使用Rich进度条来显示任务的进度。

此刻，我分两个步骤进行操作：

• pool.submit()所有任务
• for future in as_completed(xxxx)等待下一个未来。

问题在于，第一步（pool.submit）可能要花一些时间（因为我正在走目录），并且UI也没有更新，即使将来已经可用。

因此，我试图提出一个将在我的池中提交的线程，而主线程将等待下一个Future并更新UI：

"""
Usage: walker.py [options] <file/directory>...

Options:
    -r --recursive                  Walk directories recursively
    -w WORKERS --workers=WORKERS    Specify the number of process pool workers [default: 4]
    -d --debug                      Enable debug output
    -h --help                       Display this message
"""
import os
import threading
import time
from concurrent.futures._base import as_completed
from concurrent.futures.process import ProcessPoolExecutor
from pathlib import Path
from random import randint
from typing import List

from docopt import docopt
from rich.console import Console
from rich.progress import BarColumn,Progress,TextColumn


def walk_filepath_list(filepath_list: List[Path],recursive: bool = False):
    for path in filepath_list:
        if path.is_dir() and not path.is_symlink():
            if recursive:
                for f in os.scandir(path):
                    yield from walk_filepath_list([Path(f)],recursive)
            else:
                yield from (Path(f) for f in os.scandir(path))
        elif path.is_file():
            yield path


def process_task(filepath):
    rand = randint(0,1)
    time.sleep(rand)


def thread_submit(pool,filepath_list,recursive,future_to_filepath):
    for filepath in walk_filepath_list(filepath_list,recursive):
        future = pool.submit(process_task,filepath)
        # update shared dict
        future_to_filepath[future] = filepath


def main(args):
    filepath_list = [Path(entry) for entry in args["<file/directory>"]]
    debug = args["--debug"]
    workers = int(args["--workers"])
    recursive = args["--recursive"]

    console = Console()

    process_bar = Progress(
        TextColumn("[bold blue]Processing...",justify="left"),BarColumn(bar_width=None),"{task.completed}/{task.total}","•","[progress.percentage]{task.percentage:>3.1f}%",console=console,)
    process_bar.start()

    # we need to consume the iterator once to get the total
    # for the progress bar
    count = sum(1 for i in walk_filepath_list(filepath_list,recursive))
    task_process_bar = process_bar.add_task("Main task",total=count)
    with ProcessPoolExecutor(max_workers=workers) as pool:
        # shared dict between threads
        # [Future] => [filepath]
        future_to_filepath = {}
        submit_thread = threading.Thread(
            target=thread_submit,args=(pool,future_to_filepath)
        )
        submit_thread.start()
        while len(future_to_filepath.keys()) != count:
            for future in as_completed(future_to_filepath):
                filepath = future_to_filepath[future]
                # print(f"processing future: {filepath}")
                try:
                    data = future.result()
                finally:
                    # update progress bar
                    process_bar.update(task_process_bar,advance=1)
    process_bar.stop()


def entrypoint():
    args = docopt(__doc__)
    main(args)


if __name__ == "__main__":
    entrypoint()

但是，进度栏未按预期更新。 更糟糕的是，在某些情况下处理似乎没有结束。

• 我更新字典future_to_filepath时是比赛条件吗？
• 您将如何使用concurrent.futures创建一个 submit 线程和一个 process_results 线程？

非常感谢！

解决方法

查看我对您问题的评论，然后：

更改：

submit_thread = threading.Thread(
    target=thread_submit,args=(pool,filepath_list,recursive,future_to_filepath)
)
submit_thread.start()

收件人：

thread_submit(pool,future_to_filepath)

（对此函数名称进行更改，因为它不再作为单独的线程运行，这是件好事-create_futures呢？）

并删除外部循环：

while len(future_to_filepath.keys()) != count:

最后，不清楚您的实际process_task将如何处理该文件，但是肯定有可能受I / O约束。在这种情况下，您可能会受益于使用ThreadPoolExecutor类，而该类可以很容易地替换为ProcessPoolExecutor类，在这种情况下，您应该考虑指定更多的工人，可能等于{{1} }。由于您当前的count除了睡觉外没有其他事情，因此可能会通过与更多的工人一起穿线而受益。

更新

您可以减少运行process_task 的时间的一件事是将函数修改为传递单个路径而不是列表，并处理每个路径在原始列表中同时存在于单独的线程中。在下面的代码中，为了方便起见，我使用walk_filepath_list ThreadPoolExecutor函数，这实际上要求对（新重命名的）map函数的参数进行反转，以便可以使用{{1} }以“编码”所有调用的第一个参数walk_filepath：

functools.partial

更新2

上述代码的基准测试表明，它不能节省时间（也许目录是否位于不同的物理驱动器上？）。