在 ipython 并行中启动新任务时的通知

Question

几个与 ipyparallel 最佳实践相关的问题。我正在尝试使用它来为需要约 15 次运行的模型实现蒙特卡洛框架。这个想法是运行 N 个引擎（通过 SLURM）并拥有一个“主”进程，该进程将所有必需的任务异步排队并忙于等待完成，用每次运行的状态更新一个 sqlite 数据库。

我想知道何时将任务分配给引擎，以便在我的数据库中跟踪其状态。我尝试使用 AsyncResult 实例来获取 msg_id 并查询任务数据库，但是在任务完成之前，“started”字段不会更新。

似乎应该有一种方法可以接收此通知，或者至少可以在引擎工作时查询集线器。

另外，在长时间运行的任务期间，我是否必须采取措施避免引擎心跳超时？这就是client.spin_thread()的目的吗？

谢谢！

Answer 1

我已经使用 publish_data 回答了我自己的部分问题。这个想法是，我不只是在每个引擎中调用我的主要“worker”函数，而是在主要工作函数之前和之后调用 publish_data() 来设置客户端可以看到的状态。例如：

def wrapper(run_id, argDict):
    from ipyparallel.engine.datapub import publish_data

    publish_data({run_id : 'running'})
    status = runMonteCarloTrial(argDict)   # runs for ~15 minutes
    publish_data({run_id : status})
    return status

“主”任务调用：

ar = client.map_async(wrapper, listOfArgDicts) 

然后我遍历ar 直到所有AsyncResults 都完成，检查ar.data 以读取发布的数据以识别正在运行的试验并将试验结果保存到sqlite3 数据库。

这种通用方法适用于简单的测试用例。我还没有研究过长时间运行的函数调用的超时问题。