ipyparallel完成状态监控

Question

我有一个函数需要一段时间才能计算，并且必须使用两个不同的参数迭代 >20k 次：

from ipyparallel import Client
import numpy as np

m_array = np.arange(0, 101, 1)
s_array = np.arange(0, 201, 1)
rc = Client()
rc[:].push(dict(stuff=stuff))
view = rc.load_balanced_view()
async_results = []

for m in m_array:
    for s in s_array:
        chi = view.apply_async(run_simulation, m=m, s=s)
        async_results.append(chi)
rc.wait(async_results)
results = [ar.get() for ar in async_results]

我看到有一个 wait_interactive 方法可用，但我还不知道如何使用它。在给定的时间间隔打印状态更新的最佳方法是什么？

更新

我添加了 all_ids 列表和 get_result().wait_interative() 方法。

async_results = []
all_ids = []
for m in m_array:
    for s in s_array:
        chi = view.apply_async(run_simulation, m=m, s=s)
        async_results.append(chi)
        all_ids.extend(chi.msg_ids)
rc.get_result(all_ids).wait_interactive()
rc.wait(async_results)
results = [ar.get() for ar in async_results]

这会按预期生成定期状态更新，但现在会生成回溯。

---------------------------------------------------------------------------
KeyError                                  Traceback (most recent call last)
<ipython-input-36-85db6ca605cd> in <module>()
    220 rc.get_result(all_ids).wait_interactive()
    221 rc.wait(async_results)
--> 222 results = [ar.get() for ar in async_results]
223 results = np.array(results)
224 results.shape = (len(m_array), len(s_array))

//anaconda/lib/python2.7/site-packages/ipyparallel/client/asyncresult.pyc in get(self, timeout)
     95         by get() inside a `RemoteError`.
     96         """
---> 97         if not self.ready():
     98             self.wait(timeout)
     99 

//anaconda/lib/python2.7/site-packages/ipyparallel/client/asyncresult.pyc in ready(self)
    113         """Return whether the call has completed."""
    114         if not self._ready:
--> 115             self.wait(0)
    116         elif not self._outputs_ready:
    117             self._wait_for_outputs(0)

//anaconda/lib/python2.7/site-packages/ipyparallel/client/asyncresult.pyc in     wait(self, timeout)
    152                 if self.owner:
    153 
--> 154                     self._metadata = [self._client.metadata.pop(mid) for mid in self.msg_ids]
155                     [self._client.results.pop(mid) for mid in self.msg_ids]
    156 

KeyError: '884328c8-d768-48d5-b477-a256ebaea7a9'

消息 ID 或结果是否在 ar.get() 方法找到它们之前被清除了？

Answer 1

wait_interactive 是 AsyncResult 对象上的一种方法。它很快就会成为客户端本身的一种方法，但目前还没有。这意味着要使用 wait_interactive，您需要构建一个包含所有结果的 AsyncResult。最简单的方法是维护与您的请求相对应的所有 msg_ids 的单个列表：

all_ids = []
for m in m_array:
    for s in s_array:
        chi = view.apply_async(run_simulation, m=m, s=s)
        async_results.append(chi)
        all_ids.extend(chi.msg_ids)

rc.get_result(all_ids, owner=False).wait_interactive()

ipyparallel完成状态监控

Monitoring status of ipyparallel completion

ipython-notebook

ipython-parallel