Asynchronous evaluation in ipython parallel

问题:

Since the awesome 1.0.0 release I've been playing around with iPython parallel interface. What I'm trying to do is to set up a asynchronous stochastic gradient descent system. The way I see it, I want to send a function to all the nodes and get the results as they come out. From what I was able to implement and glance from the documentation the standard views implemented don't really support that. The get(timeout) method would do that, but you can't really loop through every entry in a <ASync_result> object using a timeout. The way I managed to get it running was the following

c = Client()
calls = []
for i,j in enumerate(args):
    calls.append( c[ i % len( c.ids ) ].apply( f, j ) )

while condition:
    dels = []
    for i,j in enumerate( calls ):
         try:
             print j.get(0.01) #or some other timeout
             dels.append( i ) #I keep track of the calls that have been called
             #do something with the last result, throw a new call
             calls.append( c[ i % len(c.ids) ].apply( f, argument )
         except:
             pass

    for i,d in enumerate( dels ):
         del calls[ d - i ] #delete gotten calls

    #evaluate stopping condition

Now, before you all go screaming that this is horrible code and a stupid way to do that, I know it. I could make this particular way of doing it nicer, but I'm just wondering if there is some built-in way of doing something similar in IPython.parallel.

Thanks in advance to anyone taking the time.

Best, Al.

回答1:

You can create multiple async calls, and then iterate through them.

c = Client()
dview = c[:]
asyncs = [dview.map_async(f, [arg]) for arg in args]
while asyncs:
    for async in asyncs[:]:
        if async.ready():
            asyncs.remove(async)
            print async.result[0]