title: asyncio run multiple tasks while preserving order and handling errors
body: <p>my function <code>run_tasks(all_tasks, window_size)</code> that takes a generator of <a href="https://docs.python.org/3/library/asyncio.html" rel="nofollow noreferrer"><code>asyncio</code></a> tasks and returns their value while:</p>
  
  <ol>
  <li>run each <strong><em>window</em></strong> (of size <code>window_size</code>) from the <code>all_tasks</code> concurrently</li>
  <li>preserve the order of returned results (<code>all_tasks[i]</code> result is <code>results[i]</code>)</li>
  <li>handle exceptions for each run</li>
  </ol>
  
  <p>My current implementation:</p>
  
  <pre><code>import asyncio
  from itertools import islice
  
  
  # run all tasks and return their results in the same order
  # window is the max number of tasks that will run in parallel
  def run_tasks(all_tasks, window_size=4):
      loop = asyncio.get_event_loop()
  
      while True:
          window_tasks = list(islice(all_tasks, window_size))
          if not window_tasks:
              break
  
          futures = asyncio.wait(window_tasks, loop=loop)
          finished, unfinished = loop.run_until_complete(futures)
  
          # sort finished tasks by their launch order.
          # removing this line makes returned tasks unordered
          finished = sorted(finished, key=lambda f: window_tasks.index(f._coro))
  
          for finished_task in finished:
              try:
                  yield finished_task.result()
              except Exception as e:
                  yield repr(e)
  
  # Example Usage:
  
  # a coroutine that sometime raises exception
  async def sleepy(i):
      print(f'{i} started')
      await asyncio.sleep(10 - i)
      print(f'{i} finished')
      if i == 5:
          raise ValueError('5 is the worst')
      return i
  
  # a generator of tasks
  all_tasks = (sleepy(i) for i in range(10))
  
  for result in list(run_tasks(all_tasks)):
      print(result)
  </code></pre>
  
  <h1>The Problem</h1>
  
  <p>The problem with my implementation is that I cannot find a way to sort the tasks in without accessing <code>f._coro</code> which is internal property of <code>asyncio.Task</code> object.</p>
  
  <pre><code># removing this line makes returned tasks unordered
  finished = sorted(finished, key=lambda f: window_tasks.index(f._coro))
  </code></pre>
  
  <p>I can use <code>asyncio.gather(*tasks)</code> but this will not handle errors.</p>
  
  <p>I am open to suggestions on how to implement this three properties for <code>run_tasks()</code> without access to <code>f._coro</code>.</p>