为什么这种 for 循环并行化在 Python 中不起作用？

Question

我需要浏览 10,000 个文件夹，从每个文件夹中收集一些数据，将其添加到 3 个容器（c18、c17、c16、3 个最初为空的列表中，每个列表将填充 10,000 个数字），如果没有，这将需要永远并行化。

我的目标是使用 for 循环遍历所有文件夹（for i in range(10000)），并在 for 循环的每次迭代中分别将从每个文件夹中提取的 3 个值附加到 c18、c17、c16。 我还想显示一个进度条 - 大致了解需要多长时间。

我之前从未并行化循环或包含进度条。我曾尝试使用 SO。在阅读了一些答案后，我进入了我写的点：

pool = multiprocessing.Pool(4)
pool.imap(funct, tqdm.tqdm(range(len(a0s))) # or pool.map(funct, tqdm.tqdm(range(len(a0s))))

len(a0s) 产生 10,000 个。

函数 funct 是 def funct(i): 并执行我上面写的操作：对于使用 for 循环变量 i（当前迭代次数）定义的给定文件夹，它完成提取 3 个值并附加它们的工作到 c18、c17、c16。

我在 main() 函数中调用 pool.imap(funct, tqdm.tqdm(range(len(a0s)))，并在我编写的 .py 脚本末尾：

if __name__ == '__main__':
    main()

我正在导入：

import processing
import tqdm

但是，以上所有方法都不起作用。 我该如何进行？欢迎任何帮助。 谢谢！

a0s = np.loadtxt("Intensity_Wcm2_versus_a0_10_21_10_23_range.txt", usecols=(1,)) # has 10,000 entries
pool = multiprocessing.Pool(4)

top_folder_path = os.getcwd()
base_path = top_folder_path + "/a0_"

for i in range(len(a0s)):
    results_folder = base_path + "{:.4f}".format(a0s[i])
    if os.path.isdir(results_folder):
        os.chdir(results_folder)
        S = happi.Open(".")
        pbb = S.ParticleBinning(0).get() # charge states diagnostic
        c18.append(pbb['data'][-1][-1]) # first -1 is for last timestep recorded by diagnostic, second -1 is for last charge state (bare ions, Ar18+)
        c17.append(pbb['data'][-1][-2])
        c16.append(pbb['data'][-1][-2])
        print("###########################################################]#########")
        print("We have done the folder number: " + str(i) + " out of: " + str(len(a0s)))
        os.chdir(top_folder_path)

    else:
        continue
 
 def funct(i):
    results_folder = base_path + "{:.4f}".format(a0s[i])
    if os.path.isdir(results_folder):
        os.chdir(results_folder)
        S = happi.Open(".")
        pbb = S.ParticleBinning(0).get() # charge states diagnosti
        c18_val = pbb['data'][-1][-1]
        c17_val = pbb['data'][-1][-2]
        c16_val = pbb['data'][-1][-3]
        c18.append(c18_val)
        c17.append(c17_val)
        c16.append(c16_val)
    else:
        return

def main():
    pool.imap(funct, tqdm(range(len(a0s))))

if __name__ == '__main__':
    main()

Answer 1

这是一个用于多个进度条和多处理的模板。希望能帮助到你。我将其设置为期望在每个进程中更新 10 次，并添加一个睡眠作为并行化的“工作”。

import multiprocessing as mp
import tqdm
import time
from itertools import repeat

def funct(lock,i):
    with lock:
        bar = tqdm.tqdm(position=i,total=10,leave=False,ncols=100)
    bar.set_lock(lock)
    for _ in range(10):
        time.sleep(.2)
        bar.update(1)
    bar.close()
    return i*2

def main():
    lock = mp.Manager().Lock()
    with mp.Pool() as pool:
        result = pool.starmap(funct, zip(repeat(lock),range(8)))
    print()
    print(result)

if __name__ == '__main__':
    main()