2

我想在 ipython 并行映射到达时迭代一些异步结果。我能找到的唯一方法是遍历结果对象。但是，如果其中一项任务引发异常，则迭代终止。有没有办法做到这一点？请参见下面的代码，当第二个作业引发异常时，迭代终止。

from IPython import parallel

def throw_even(i):
    if i % 2 == 0:
        raise RuntimeError('ERROR: %d' % i)
    return i

rc = parallel.Client()
lview = rc.load_balanced_view() # default load-balanced view

# map onto the engines.
args = range(1, 5)
print args
async_results = lview.map_async(throw_even, range(1, 5), ordered=True)

# get results
args_iter = iter(args)
results_iter = iter(async_results)
while True:
    try:
        arg = args_iter.next()
        result = results_iter.next()
        print 'Job %s completed: %d' % (arg, result)            
    except StopIteration:
        print 'Finished iteration'
        break
    except Exception as e:
        print '%s: Job %d: %s' % (type(e), arg, e)

给出在报告作业 3 和 4 之前停止的以下输出

[1, 2, 3, 4]
Job 1 completed: 1
<class 'IPython.parallel.error.RemoteError'>: Job 2: RuntimeError(ERROR: 2)
Finished iteration

有没有办法做到这一点？

4

2 回答 2

0

解决这个问题的一个偷偷摸摸的方法是进入内部AsyncMapResult并抓取_result结果列表。这不会直接帮助您，但只能在事后帮助您：

tt = async_results._results
fail_indx = [j for j, r in enumerate(tt) if isinstance(r, IPython.parallel.error.RemoteError)]
good_indx = [j for j, r in enumerate(tt) if not isinstance(r, IPython.parallel.error.RemoteError)]

just_the_results =  [r for  r in tt if not isinstance(r, IPython.parallel.error.RemoteError)]
于 2013-11-15T04:34:33.527 回答
0

This question might be relevant. I dont't really see why you would want to throw an exception from a remote engine, though. Although, if you do want to do it, I think you can do it in the same way I answered the question mentioned. Which I now see you already realized in your comments, but this should do it anyway.

def throw_even(i):
    if i%2:
       return i
    raise(RuntimeError('Error %d'%i)

params = range(1,5)

n_cores = len(c.ids)
for n,p in enumerate( params ):
    core = c.ids[n%n_cores]
    calls.append( c[core].apply_async( throw_even, p ) )

#then you get the results

while calls != []:
    for c in calls:
        try:
             result = c.get(1e-3)
             print(result[0])
             calls.remove( c )
             #in the case your call failed, you can apply_async again.
             # and append the call to calls.
        except parallel.TimeoutError:
             pass
        except Exception as e:
             knock_yourself_out(e)
于 2013-10-23T12:42:06.560 回答