Equivalent of asyncio.gather()? about curio HOT 11 CLOSED

Running multiple coroutines in parallel should not be a problem, but there is no explicit gather() function in the first version. So, you would have to explicitly join() with each one to wait for results.

This is definitely a part of curio that could be expanded in the future. I remember looking into it awhile back, but decided to hold off in the interest of keeping things simple in an initial version. There are some more advanced waiting functions in asyncio that might be worth looking at as well (for example, the ability to wait for any task to finish).

I have this gut feeling that much of this functionality could be implemented on top of curio itself as opposed to be being dropped into the low-level kernel code. For example:

async def gather(tasks):
          results = []
          for task in tasks:
                    result = await task.join()
                    results.append(result)
          return results

Obviously, there are some more details that would need to be fleshed out with it (i.e., cancellation and exception handling).

from curio.

seems there's an undocumented curio.gather in 0.4 now:

In [7]: curio.gather?
Signature: curio.gather(tasks, *, return_exceptions=False)
Docstring: Wait for and gather results from a collection of tasks.
File:      ~/src/curio/curio/task.py
Type:      function

OTOH, a curio.race function would be nice/interesting, i.e. waiting for the first task to finish. For. ex. dns query to several servers simultaneously. ps. or first.

from curio.

Yes. Undocumented for now.

One thing I'm just not sure about is how this function should actually work--meaning the underlying semantics as well as inputs. Questions arise:

1. Should it operate on tasks (created by spawn()) or simply a collection of coroutines? If the latter, I assume that gather() would implicitly spawn them into tasks.
2. How does error handling work? If one of the tasks fails, do all of the remaining tasks get cancelled?
3. How does cancellation work? Would it cancel only the gather() operation or does it propagate and cancel all of the tasks being watched?
4. Could gather() return partial results. For example, if there was a timeout, could it return results gathered so far, but then be called again on the remaining tasks to get their results?
5. Do results have to come back in order?

Initially, I thought it might make sense to simply copy what asyncio was doing. However, I'm now no longer sure. So, with all of this said, I'd be really interest to get thoughts about how a gather() function should work. The idea of a race() or first() function would be interesting as well. It would have some of the same issues (e.g., do tasks that haven't finished get cancelled?).

from curio.

wrt race/first IMHO there should be an option to cancel them but by default no, race(*tasks, cancel_rest=False, timeout=None) …
ps. I know timeout can already be composed with other features in curio, but this seems nicer to me.

pps. the only question is, if all the tasks fail, I guess the race should fail too, ie throw some custom Exception.

from curio.

gather, can be also called all or even map, it means all tasks need to finish, return a result, and the results are in order. IMHO

from curio.

Regarding order for gather / all, I think it can sometimes be useful to get tasks back in the order they finished, not just in the order they were passed.

from curio.

I wonder if something like this could work as a context manager. Or maybe an iterator. For example::

async with gather(t1, t2, t3) as g:
     task = await g.next()        
     ...

async with gather(t1, t2, t3) as g:
     async for task in g:
           ...

from curio.

typically, the point of all/gather is to have them (to know) all finished. So the when async with gather returns you wouldn't need an async for, they should have the result.

from curio.

@jab that's what race is about. you get the first, then you can race the rest if you want.

from curio.

I guess what I'm wondering is whether or not both kinds of functionality could be consolidated into a single core feature. If you can get the tasks one at a time (race) then gather is an extension. I will mull this over a bit.

from curio.

I added a new wait() function to curio that can be used to play with some of these ideas. Here's an example::

 # Some tasks
  task1 = await spawn(coro())
  task2 = await spawn(coro())
  task3 = await spawn(coro())

  # Wait for them in completion order
  async for task in curio.wait([task1, task2, task3]):
          result = await task.join()
          print('Result ->', result)

  # Wait for the first task to complete then cancel the rest
  async with curio.wait([task1, task2, task3]) as w:
          task = await w.next_done()
          result = await task.join()
          print('Result ->', result)
  # Remaining tasks cancelled here

I might change it further. Just experimenting for now.

from curio.