You can't use View.map
with a generator without walking through the entire generator first. But you can write your own custom function to submit batches of tasks from a generator and wait for them incrementally. I don't have a more interesting example, but I can illustrate with a terrible implementation of a prime search.
Start with our token 'data generator':
from math import sqrt
def generate_possible_factors(N):
"""generator for iterating through possible factors for N
yields 2, every odd integer <= sqrt(N)
"""
if N <= 3:
return
yield 2
f = 3
last = int(sqrt(N))
while f <= last:
yield f
f += 2
This just generates a sequence of integers to use when testing if a number is prime.
Now our trivial function that we will use as a task with IPython.parallel
def is_factor(f, N):
"""is f a factor of N?"""
return (N % f) == 0
and a complete implementation of prime check using the generator and our factor function:
def dumb_prime(N):
"""dumb implementation of is N prime?"""
for f in generate_possible_factors(N):
if is_factor(f, N):
return False
return True
A parallel version that only submits a limited number of tasks at a time:
def parallel_dumb_prime(N, v, max_outstanding=10, dt=0.1):
"""dumb_prime where each factor is checked remotely
Up to `max_outstanding` factors will be checked in parallel.
Submission will halt as soon as we know that N is not prime.
"""
tasks = set()
# factors is a generator
factors = generate_possible_factors(N)
while True:
try:
# submit a batch of tasks, with a maximum of `max_outstanding`
for i in range(max_outstanding-len(tasks)):
f = factors.next()
tasks.add(v.apply_async(is_factor, f, N))
except StopIteration:
# no more factors to test, stop submitting
break
# get the tasks that are done
ready = set(task for task in tasks if task.ready())
while not ready:
# wait a little bit for some tasks to finish
v.wait(tasks, timeout=dt)
ready = set(task for task in tasks if task.ready())
for t in ready:
# get the result - if True, N is not prime, we are done
if t.get():
return False
# update tasks to only those that are still pending,
# and submit the next batch
tasks.difference_update(ready)
# check the last few outstanding tasks
for task in tasks:
if t.get():
return False
# checked all candidates, none are factors, so N is prime
return True
This submits a limited number of tasks at a time, and as soon as we know that N is not prime, we stop consuming the generator.
To use this function:
from IPython import parallel
rc = parallel.Client()
view = rc.load_balanced_view()
for N in range(900,1000):
if parallel_dumb_prime(N, view, 10):
print N
A more complete illustration in a notebook.