我正在使用Python的multiprocessing库中的starmap_async
函数。 但是,如果我的代码在某个进程中遇到错误,则不会抛出异常,直到所有进程都完成。以下是相关的代码:
from multiprocessing import Pool, cpu_count
import datetime
import itertools
import time
with Pool(max(cpu_count()//2, 1)) as p:
#p = Pool()
df_iter = df_options.iterrows()
ir = itertools.repeat
results = p.starmap_async(_run,zip(df_iter,ir(fixed_options),ir(outputs_grab)), chunksize=1)
p.close() #no more jobs to submit
#Printing progress
n_remaining = results._number_left + 1
while (not results.ready()):
time.sleep(1)
#Check for errors here ... How ????
#Then what? call terminate()?????
if verbose:
if results._number_left < n_remaining:
now = datetime.datetime.now()
n_remaining = results._number_left
print('%d/%d %s' % (n_remaining,n_rows,str(now)[11:]))
print('joining')
p.join()
all_results = results.get()
df = pd.DataFrame(all_results)
目前,如果我在生成的进程中引发错误,似乎其他进程不仅会完成运行,而且还会开始新任务,尽管其中一个调用出现错误。
一些搜索使我相信这可能是不可能的。有人似乎建议我改用 concurrent.futures
,尽管如何将我的示例映射到该示例仍不清楚,特别是在保持实时反馈完成进程的同时。
concurrent.futures
的讨论: https://dev59.com/9mw15IYBdhLWcg3wQ5di#47108581
pool.apply_async
逐个发送它们(无阻塞),这样你就可以像pool.imap_unordered
一样跟踪结果的到来。 - Charchit Agarwalimap_unordered
来最小化延迟(主进程不等待正确的有序输出来生成下一个循环迭代),但常规的imap
也可以保持顺序,并基于chunksize
和工作器数量给出给定的最小延迟。 - Aaronraise
即可。我还添加了一个枚举,以便在进度进行时可以获得计数器以打印进度。谢谢! - Jimbo