src/black/concurrency.py

   1 import asyncio
   2 import logging
   3 import sys
   4 from typing import Any, Iterable
   5
   6 from black.output import err
   7
   8
   9 def cancel(tasks: Iterable["asyncio.Task[Any]"]) -> None:
  10     """asyncio signal handler that cancels all `tasks` and reports to stderr."""
  11     err("Aborted!")
  12     for task in tasks:
  13         task.cancel()
  14
  15
  16 def shutdown(loop: asyncio.AbstractEventLoop) -> None:
  17     """Cancel all pending tasks on `loop`, wait for them, and close the loop."""
  18     try:
  19         if sys.version_info[:2] >= (3, 7):
  20             all_tasks = asyncio.all_tasks
  21         else:
  22             all_tasks = asyncio.Task.all_tasks
  23         # This part is borrowed from asyncio/runners.py in Python 3.7b2.
  24         to_cancel = [task for task in all_tasks(loop) if not task.done()]
  25         if not to_cancel:
  26             return
  27
  28         for task in to_cancel:
  29             task.cancel()
  30         loop.run_until_complete(
  31             asyncio.gather(*to_cancel, loop=loop, return_exceptions=True)
  32         )
  33     finally:
  34         # `concurrent.futures.Future` objects cannot be cancelled once they
  35         # are already running. There might be some when the `shutdown()` happened.
  36         # Silence their logger's spew about the event loop being closed.
  37         cf_logger = logging.getLogger("concurrent.futures")
  38         cf_logger.setLevel(logging.CRITICAL)
  39         loop.close()