title: Wrapping asyncio.gather in a timeout
body: <p>I've seen <a href="https://stackoverflow.com/q/42231161/7954504">asyncio.gather vs asyncio.wait</a>, but am not sure if that addresses this particular question.  What I'm looking to do is wrap the <code>asyncio.gather()</code> coroutine in <code>asyncio.wait_for()</code>, with a <code>timeout</code> argument.  I also need to satisfy these conditions:</p>
  
  <ul>
  <li><code>return_exceptions=True</code> (from <code>asyncio.gather()</code>) - rather than propagating exceptions to the task that awaits on <code>gather()</code>, I want to include exceptions instances in the results</li>
  <li>Order: retain the property of <code>asyncio.gather()</code> that the order of results is the same as the order of the input.  (Or otherwise, map the output back to the input.).  <code>asyncio.wait_for()</code> fails this criteria and I'm not sure of ideal way to achieve it.</li>
  </ul>
  
  <p>The timeout is for the <em>entire</em> <code>asyncio.gather()</code> across the list of awaitables--if they get caught in the timeout or return an exception, either of those cases should just place an exception instance in the result list.</p>
  
  <p>Consider this setup:</p>
  
  <pre><code>&gt;&gt;&gt; import asyncio
  &gt;&gt;&gt; import random
  &gt;&gt;&gt; from time import perf_counter
  &gt;&gt;&gt; from typing import Iterable
  &gt;&gt;&gt; from pprint import pprint
  &gt;&gt;&gt; 
  &gt;&gt;&gt; async def coro(i, threshold=0.4):
  ...     await asyncio.sleep(i)
  ...     if i &gt; threshold:
  ...         # For illustration's sake - some coroutines may raise,
  ...         # and we want to accomodate that and just test for exception
  ...         # instances in the results of asyncio.gather(return_exceptions=True)
  ...         raise Exception("i too high")
  ...     return i
  ... 
  &gt;&gt;&gt; async def main(n, it: Iterable):
  ...     res = await asyncio.gather(
  ...         *(coro(i) for i in it),
  ...         return_exceptions=True
  ...     )
  ...     return res
  ... 
  &gt;&gt;&gt; 
  &gt;&gt;&gt; random.seed(444)
  &gt;&gt;&gt; n = 10
  &gt;&gt;&gt; it = [random.random() for _ in range(n)]
  &gt;&gt;&gt; start = perf_counter()
  &gt;&gt;&gt; res = asyncio.run(main(n, it=it))
  &gt;&gt;&gt; elapsed = perf_counter() - start
  &gt;&gt;&gt; print(f"Done main({n}) in {elapsed:0.2f} seconds")  # Expectation: ~1 seconds
  Done main(10) in 0.86 seconds
  &gt;&gt;&gt; pprint(dict(zip(it, res)))
  {0.01323751590501987: 0.01323751590501987,
   0.07422124156714727: 0.07422124156714727,
   0.3088946587429545: 0.3088946587429545,
   0.3113884366691503: 0.3113884366691503,
   0.4419557492849159: Exception('i too high'),
   0.4844375347808497: Exception('i too high'),
   0.5796792804615848: Exception('i too high'),
   0.6338658027451068: Exception('i too high'),
   0.7426396870165088: Exception('i too high'),
   0.8614799253779063: Exception('i too high')}
  </code></pre>
  
  <p>The program above, with <code>n = 10</code>, has an exected runtime of .5 seconds plus a bit of overhead when run asynchronously.  (<code>random.random()</code> will be uniformly distributed in [0, 1).)</p>
  
  <p>Let's say I want to impose that as the timeout, on the entire operation (i.e. on the coroutine <code>main()</code>):</p>
  
  <pre><code>timeout = 0.5
  </code></pre>
  
  <p>Now, I can use <code>asyncio.wait()</code>, but the problem is that the results are <code>set</code> objects and so definitely can't guarantee the sorted return value property of <code>asyncio.gather()</code>:</p>
  
  <pre><code>&gt;&gt;&gt; async def main(n, it, timeout) -&gt; tuple:
  ...     tasks = [asyncio.create_task(coro(i)) for i in it]
  ...     done, pending = await asyncio.wait(tasks, timeout=timeout)
  ...     return done, pending
  ... 
  &gt;&gt;&gt; timeout = 0.5
  &gt;&gt;&gt; random.seed(444)
  &gt;&gt;&gt; it = [random.random() for _ in range(n)]
  &gt;&gt;&gt; start = perf_counter()
  &gt;&gt;&gt; done, pending = asyncio.run(main(n, it=it, timeout=timeout))
  &gt;&gt;&gt; for i in pending:
  ...     i.cancel()
  &gt;&gt;&gt; elapsed = perf_counter() - start
  &gt;&gt;&gt; print(f"Done main({n}) in {elapsed:0.2f} seconds")
  Done main(10) in 0.50 seconds
  &gt;&gt;&gt; done
  {&lt;Task finished coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt; exception=Exception('i too high')&gt;, &lt;Task finished coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt; exception=Exception('i too high')&gt;, &lt;Task finished coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt; result=0.3088946587429545&gt;, &lt;Task finished coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt; result=0.3113884366691503&gt;, &lt;Task finished coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt; result=0.01323751590501987&gt;, &lt;Task finished coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt; result=0.07422124156714727&gt;}
  &gt;&gt;&gt; pprint(done)
  {&lt;Task finished coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt; exception=Exception('i too high')&gt;,
   &lt;Task finished coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt; result=0.3113884366691503&gt;,
   &lt;Task finished coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt; result=0.07422124156714727&gt;,
   &lt;Task finished coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt; exception=Exception('i too high')&gt;,
   &lt;Task finished coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt; result=0.01323751590501987&gt;,
   &lt;Task finished coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt; result=0.3088946587429545&gt;}
  &gt;&gt;&gt; pprint(pending)
  {&lt;Task cancelled coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt;&gt;,
   &lt;Task cancelled coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt;&gt;,
   &lt;Task cancelled coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt;&gt;,
   &lt;Task cancelled coro=&lt;coro() done, defined at &lt;stdin&gt;:1&gt;&gt;}
  </code></pre>
  
  <p>As stated above, the issue is that I seemingly can't map back <code>task</code> instances to the inputs in <code>iterable</code>.  They task ids are effectively lost inside a function scope with <code>tasks = [asyncio.create_task(coro(i)) for i in it]</code>.  Is there a Pythonic way/use of asyncio API to mimic the behavior of <code>asyncio.gather()</code> here?</p>