title: asyncio.wait_for_completion() on dynamic list of tasks
body: <p>I am writing a simple client to interact with a possibly overloaded and unreliable webserver. I assume that for any individual request, the server may never respond (request will timeout), or may respond with an error after a lengthy delay.</p>
  <p>Because of this, for each &quot;request&quot; I want to issue repeated requests according to the following logic:</p>
  <ul>
  <li>Repeat the request (issue a new request) every second until one of the issued requests receives the expected response</li>
  <li>New requests should be issued concurrently with existing requests - existing requests should stay open until they receive a response or time out.</li>
  <li>Once the expected response is received, cancel (close) any requests that are still pending/open.</li>
  </ul>
  <p>I can accomplish something close to this if I issue a fixed number of requests simultaneously at the beginning, then use <code>asyncio.as_completed()</code> to handle the requests as they finish and cancel any remaining pending requests:</p>
  <pre class="lang-py prettyprint-override"><code>import asyncio
  import logging
  import random
  import time
  from sys import stdout
  
  
  class FailedRequest(Exception):
      pass
  
  async def get():
      '''A simple mock async GET request that returns a randomized status after a randomized delay'''
      await asyncio.sleep(random.uniform(0,10))
      return random.choices([200, 500], [0.2, 0.8])[0]
  
  async def fetch(id):
      '''Makes a request using get(), checks response, and handles cancellation'''
      logging.info(f&quot;Sending request {id}.&quot;)
      start_time = time.perf_counter()
      try:
          response = await get()
          elapsed_time = time.perf_counter() - start_time
              
          if response != 200:
              logging.error(f&quot;Request {id} failed after {elapsed_time:.2f}s: {response}&quot;)
              raise FailedRequest()
          else:
              logging.info(f&quot;Request {id} succeeded ({response}) after {elapsed_time:.2f}s!&quot;)
      except asyncio.CancelledError:
          logging.info(f&quot;Cancelled request {id} after {time.perf_counter() - start_time:.2f}s.&quot;)
          raise
  
  async def main():
      # Create 10 unique Tasks that wrap the fetch() coroutine
      tasks = [asyncio.create_task(fetch(i)) for i in range(10)]
  
      # Iterate through the tasks as they are completed
      for coro in asyncio.as_completed(tasks):
          try:
              # Wait for the next task to finish. If the request errored out,
              # this line will raise a FailedRequest exception (caught below)
              await coro
  
              # If we get here, then a request succeeded. Cancel all of the tasks we started.
              for t in tasks:
                  t.cancel()
  
          except (FailedRequest, asyncio.CancelledError) as e:
              pass
  
      logging.info(&quot;Finished!&quot;)
                  
  if __name__ == '__main__':
      logging.basicConfig(stream=stdout, level=logging.INFO, format='%(asctime)s:%(levelname)s: %(message)s')
      random.seed(3)
      asyncio.run(main())
  </code></pre>
  <p>Output:</p>
  <pre><code>2020-09-22 18:07:35,634:INFO: Sending request 0.
  2020-09-22 18:07:35,635:INFO: Sending request 1.
  2020-09-22 18:07:35,635:INFO: Sending request 2.
  2020-09-22 18:07:35,635:INFO: Sending request 3.
  2020-09-22 18:07:35,636:INFO: Sending request 4.
  2020-09-22 18:07:35,636:INFO: Sending request 5.
  2020-09-22 18:07:35,636:INFO: Sending request 6.
  2020-09-22 18:07:35,636:INFO: Sending request 7.
  2020-09-22 18:07:35,636:INFO: Sending request 8.
  2020-09-22 18:07:35,637:INFO: Sending request 9.
  2020-09-22 18:07:35,786:ERROR: Request 6 failed after 0.15s: 500
  2020-09-22 18:07:36,301:ERROR: Request 5 failed after 0.66s: 500
  2020-09-22 18:07:37,993:ERROR: Request 9 failed after 2.35s: 500
  2020-09-22 18:07:38,023:ERROR: Request 0 failed after 2.39s: 500
  2020-09-22 18:07:38,236:ERROR: Request 8 failed after 2.60s: 500
  2020-09-22 18:07:39,351:INFO: Request 2 succeeded (200) after 3.72s!
  2020-09-22 18:07:39,351:INFO: Cancelled request 1 after 3.72s.
  2020-09-22 18:07:39,351:INFO: Cancelled request 3 after 3.72s.
  2020-09-22 18:07:39,352:INFO: Cancelled request 4 after 3.72s.
  2020-09-22 18:07:39,352:INFO: Cancelled request 7 after 3.72s.
  2020-09-22 18:07:39,352:INFO: Finished!
  </code></pre>
  <p>However, I am struggling to understand a clean way to start by issuing a single request, then issue additional requests every second until one of the requests is successful, while still keeping track of all unfinished requests and cancelling any that are still pending.</p>
  <p>This is as close as I've gotten:</p>
  <pre class="lang-py prettyprint-override"><code>import asyncio
  import logging
  import random
  import time
  from sys import stdout
  
  
  class FailedRequest(Exception):
      pass
  
  async def get():
      '''A simple mock async GET request that returns a randomized status after a randomized delay'''
      await asyncio.sleep(random.uniform(0,10))
      return random.choices([200, 500], [0.2, 0.8])[0]
  
  async def fetch(id):
      '''Makes a request using get(), checks response, and handles cancellation'''
      logging.info(f&quot;Sending request {id}.&quot;)
      start_time = time.perf_counter()
      try:
          response = await get()
          elapsed_time = time.perf_counter() - start_time
              
          if response != 200:
              logging.error(f&quot;Request {id} failed after {elapsed_time:.2f}s: {response}&quot;)
              raise FailedRequest()
          else:
              logging.info(f&quot;Request {id} succeeded ({response}) after {elapsed_time:.2f}s!&quot;)
      except asyncio.CancelledError:
          logging.info(f&quot;Cancelled request {id} after {time.perf_counter() - start_time:.2f}s.&quot;)
          raise
  
  async def issue_requests(finished, requests):
      i = 0
      while not finished.is_set():
          requests.add(asyncio.create_task(fetch(i)))
          await asyncio.sleep(1)
          i += 1
  
  async def handle_requests(finished, requests):
      # Iterate through the requests as they are completed
      for coro in asyncio.as_completed(requests):
          try:
              # Wait for the next task to finish. If the request errored out,
              # this line will raise a FailedRequest exception (caught below)
              await coro
  
              # If we get here, then a request succeeded. Cancel all of the tasks we started.
              finished.set()
              for r in requests:
                  r.cancel()
  
          except (FailedRequest, asyncio.CancelledError):
              pass
  
  
  async def main():
      finished = asyncio.Event()
      requests = set()
  
      await asyncio.gather(issue_requests(finished, requests), handle_requests(finished, requests))
      logging.info(&quot;Finished!&quot;)
                  
  if __name__ == '__main__':
      logging.basicConfig(stream=stdout, level=logging.INFO, format='%(asctime)s:%(levelname)s: %(message)s')
      random.seed(3)
      asyncio.run(main())
  </code></pre>
  <p>However, although the requests are launched as expected, the process does not stop when the first successful request returns:</p>
  <pre><code>2020-09-22 18:03:38,256:INFO: Sending request 0.
  2020-09-22 18:03:39,264:INFO: Sending request 1.
  2020-09-22 18:03:40,265:INFO: Sending request 2.
  2020-09-22 18:03:40,643:ERROR: Request 0 failed after 2.39s: 500
  2020-09-22 18:03:41,281:INFO: Sending request 3.
  2020-09-22 18:03:42,281:INFO: Sending request 4.
  2020-09-22 18:03:42,948:INFO: Request 4 succeeded (200) after 0.67s!
  # requests 1, 2, and 3 should be cancelled here and the script should finish
  2020-09-22 18:03:43,279:INFO: Sending request 5.
  2020-09-22 18:03:43,976:ERROR: Request 2 failed after 3.71s: 500
  2020-09-22 18:03:44,281:INFO: Sending request 6.
  2020-09-22 18:03:44,718:ERROR: Request 1 failed after 5.45s: 500
  2020-09-22 18:03:45,295:INFO: Sending request 7.
  2020-09-22 18:03:46,307:INFO: Sending request 8.
  ...
  </code></pre>
  <p>I think the issue is that when <code>asyncio.as_completed(requests)</code> is called in <code>handle_requests()</code>, <code>requests</code> is an empty set, so <code>as_completed()</code> returns an empty iterator and <code>handle_requests()</code> returns immediately.</p>
  <p>It feels like it should be possible to do this with <code>asyncio</code> at a high level, but I am struggling to figure it out.</p>