Feature request: Generators in processes

[noxdafox]

Feature request: Generators in processes

[noxdafox]

Are you aiming to reach any parallelism in here or is just for isolation purposes? In other words: is this a sequential operation ran in another process or do you need to split it into parallel computations?

[andresriancho]

Both isolation and parallelism. Run a browser with custom JS, load an arbitrary site / page, extract data from it. The browser might crash (isolation). Multiple browsers will run faster than just one (rendering html is cpu bound)

[noxdafox]

I see what you mean and I think it's a pretty interesting feature. I am currently deep into another thing so I will start working on this a bit later (beginning of June?). Ofc pull requests are welcome.

In the meantime, you can use this code snippet as a reference for implementing such functionality in your application.

from pebble import concurrent
from multiprocessing import Queue

@concurrent.process(timeout=60)
def process_url(queue, url):
    """Processes a URL splitting the report in chunks and putting it in the given Queue."""
    for chunk in process(url):
        queue.put(chunk)

def process(url):
    """A function seemingly producing large data split in chunks."""
    while True:
        yield None
       
queue = Queue()
future = process_url(queue, "www.example.com")

while not future.done():
    chunk = queue.get()

if future.exception() is not None:
    print("Error while processing the URL")

[andresriancho]

Ah! I like the trick of using a queue!

The example code looks good and is a good inspiration for writing the PR. Depending on my client's needs, I might be writing the PR before June.

What about the interface? Do you think we should have two new methods in ProcessPool?

• schedule_generator()
• map_generator()

Both of them returning a tuple with future and generator? The generator would be a wrapper around the queue that people could use like this:

future, generator = pool.schedule_generator(function)

for result in generator:
    print(result)

# Do some future error handling here