title: How to make a asyncio pool cancelable?
body: <p>I have a <code>pool_map</code> function that can be used to limit the number of simultaneously executing functions.</p>
  
  <p>The idea is to have a <a href="https://docs.python.org/3/library/asyncio-task.html#coroutine" rel="nofollow noreferrer">coroutine function</a> accepting a single parameter that is mapped to a list of possible parameters, but to also wrap all function calls into a semaphore acquisition, whereupon only a limited number is running at once:</p>
  
  <pre><code>from typing import Callable, Awaitable, Iterable, Iterator
  from asyncio import Semaphore
  
  A = TypeVar('A')
  V = TypeVar('V')
  
  async def pool_map(
      func: Callable[[A], Awaitable[V]],
      arg_it: Iterable[A],
      size: int=10
  ) -&gt; Generator[Awaitable[V], None, None]:
      """
      Maps an async function to iterables
      ensuring that only some are executed at once.
      """
      semaphore = Semaphore(size)
  
      async def sub(arg):
          async with semaphore:
              return await func(arg)
  
      return map(sub, arg_it)
  </code></pre>
  
  <p>I modified and didn’t test above code for the sake of an example, but my variant works well. E.g. you can use it like this:</p>
  
  <pre><code>from asyncio import get_event_loop, coroutine, as_completed
  from contextlib import closing
  
  URLS = [...]
  
  async def run_all(awaitables):
      for a in as_completed(awaitables):
          result = await a
          print('got result', result)
  
  async def download(url): ...
  
  
  if __name__ != '__main__':
      pool = pool_map(download, URLS)
  
      with closing(get_event_loop()) as loop:
          loop.run_until_complete(run_all(pool))
  </code></pre>
  
  <p>But a problem arises if there is an exception thrown while awaiting a future. I can’t see how to cancel all scheduled or still-running tasks, neither the ones still waiting for the semaphore to be acquired.</p>
  
  <p>Is there a library or an elegant building block for this that I don’t know, or do I have to build all parts myself? (i.e. a <code>Semaphore</code> with access to its waiters, a <code>as_finished</code> that provides access to its running task queue, …)</p>