def _cancel_all_tasks(loop: asyncio.AbstractEventLoop) -> None:
    tasks = [task for task in asyncio.all_tasks(loop) if not task.done()]
    if not tasks:
        return

    for task in tasks:
        task.cancel()
    loop.run_until_complete(asyncio.gather(*tasks, loop=loop, return_exceptions=True))

    for task in tasks:
        if not task.cancelled() and task.exception() is not None:
            loop.call_exception_handler(
                {
                    "message": "unhandled exception during shutdown",
                    "exception": task.exception(),
                    "task": task,
                }
            ) 

def __init__(self, io_loop: asyncio.AbstractEventLoop = None):
        super().__init__()
        self.io_loop = io_loop or asyncio.get_event_loop()
        self.sub_client = self.io_loop.run_until_complete(
                aioredis.create_redis((config.get('REDIS', 'host', fallback='localhost'),
                                       config.getint('REDIS', 'port', fallback=6379)),
                                      db=config.getint('REDIS', 'db', fallback=1)))
        self.redis_client = redis.StrictRedis(
            host=config.get('REDIS', 'host', fallback='localhost'),
            db=config.getint('REDIS', 'db', fallback=1), decode_responses=True)
        self.initialized = False
        self.sub_tasks = list()
        self.sub_channels = list()
        self.channel_router = dict()
        self.crontab_router = defaultdict(dict)
        self.datetime = None
        self.time = None
        self.loop_time = None 

def loop(self: 'TelegramClient') -> asyncio.AbstractEventLoop:
        """
        Property with the ``asyncio`` event loop used by this client.

        Example
            .. code-block:: python

                # Download media in the background
                task = client.loop.create_task(message.download_media())

                # Do some work
                ...

                # Join the task (wait for it to complete)
                await task
        """
        return self._loop 

def __init__(self, loop: asyncio.AbstractEventLoop, **kwargs):
        auth_headers = {
            'Authorization': f"Token {Keys.site_api}"
        }

        if 'headers' in kwargs:
            kwargs['headers'].update(auth_headers)
        else:
            kwargs['headers'] = auth_headers

        self.session = None
        self.loop = loop

        self._ready = asyncio.Event(loop=loop)
        self._creation_task = None
        self._default_session_kwargs = kwargs

        self.recreate() 

def handle_loop_error(
    root_app: web.Application,
    loop: asyncio.AbstractEventLoop,
    context: Mapping[str, Any],
) -> None:
    if isinstance(loop, aiojobs.Scheduler):
        loop = current_loop()
    exception = context.get('exception')
    msg = context.get('message', '(empty message)')
    if exception is not None:
        if sys.exc_info()[0] is not None:
            log.exception('Error inside event loop: {0}', msg)
            if 'error_monitor' in root_app:
                loop.create_task(root_app['error_monitor'].capture_exception())
        else:
            exc_info = (type(exception), exception, exception.__traceback__)
            log.error('Error inside event loop: {0}', msg, exc_info=exc_info)
            if 'error_monitor' in root_app:
                loop.create_task(root_app['error_monitor'].capture_exception(exception)) 

def __init__(
        self,
        token=None,
        base_url=BASE_URL,
        timeout=30,
        loop: Optional[asyncio.AbstractEventLoop] = None,
        ssl=None,
        proxy=None,
        run_async=False,
        use_sync_aiohttp=False,
        session=None,
        headers: Optional[dict] = None,
    ):
        self.token = None if token is None else token.strip()
        self.base_url = base_url
        self.timeout = timeout
        self.ssl = ssl
        self.proxy = proxy
        self.run_async = run_async
        self.use_sync_aiohttp = use_sync_aiohttp
        self.session = session
        self.headers = headers or {}
        self._logger = logging.getLogger(__name__)
        self._event_loop = loop 

def __init__(
        self, token: CancelToken = None, loop: asyncio.AbstractEventLoop = None
    ) -> None:
        self.events = ServiceEvents()
        self._run_lock = asyncio.Lock()
        self._child_services = WeakSet()
        self._tasks = WeakSet()
        self._finished_callbacks = []

        self._loop = loop

        base_token = CancelToken(type(self).__name__, loop=loop)

        if token is None:
            self.cancel_token = base_token
        else:
            self.cancel_token = base_token.chain(token) 

def __init__(self,
                 impl:           SubscriberImpl[MessageClass],
                 loop:           asyncio.AbstractEventLoop,
                 queue_capacity: typing.Optional[int]):
        """
        Do not call this directly! Use :meth:`Presentation.make_subscriber`.
        """
        if queue_capacity is None:
            queue_capacity = 0      # This case is defined by the Queue API. Means unlimited.
        else:
            queue_capacity = int(queue_capacity)
            if queue_capacity < 1:
                raise ValueError(f'Invalid queue capacity: {queue_capacity}')

        self._closed = False
        self._impl = impl
        self._loop = loop
        self._maybe_task: typing.Optional[asyncio.Task[None]] = None
        self._rx: _Listener[MessageClass] = _Listener(asyncio.Queue(maxsize=queue_capacity, loop=loop))
        impl.add_listener(self._rx)

    # ----------------------------------------  HANDLER-BASED API  ---------------------------------------- 

def __init__(self,
                 dtype:                            typing.Type[ServiceClass],
                 input_transport_session:          pyuavcan.transport.InputSession,
                 output_transport_session_factory: OutputTransportSessionFactory,
                 finalizer:                        TypedSessionFinalizer,
                 loop:                             asyncio.AbstractEventLoop):
        """
        Do not call this directly! Use :meth:`Presentation.get_server`.
        """
        self._dtype = dtype
        self._input_transport_session = input_transport_session
        self._output_transport_session_factory = output_transport_session_factory
        self._finalizer = finalizer
        self._loop = loop

        self._output_transport_sessions: typing.Dict[int, pyuavcan.transport.OutputSession] = {}
        self._maybe_task: typing.Optional[asyncio.Task[None]] = None
        self._closed = False
        self._send_timeout = DEFAULT_SERVICE_REQUEST_TIMEOUT

        self._served_request_count = 0
        self._deserialization_failure_count = 0
        self._malformed_request_count = 0

    # ----------------------------------------  MAIN API  ---------------------------------------- 

def __init__(self,
                 specifier:        pyuavcan.transport.InputSessionSpecifier,
                 payload_metadata: pyuavcan.transport.PayloadMetadata,
                 loop:             asyncio.AbstractEventLoop,
                 finalizer:        typing.Callable[[], None]):
        self._specifier = specifier
        self._payload_metadata = payload_metadata
        self._loop = loop
        self._maybe_finalizer: typing.Optional[typing.Callable[[], None]] = finalizer
        assert isinstance(self._specifier, pyuavcan.transport.InputSessionSpecifier)
        assert isinstance(self._payload_metadata, pyuavcan.transport.PayloadMetadata)
        assert isinstance(self._loop, asyncio.AbstractEventLoop)
        assert callable(self._maybe_finalizer)

        self._transfer_id_timeout = self.DEFAULT_TRANSFER_ID_TIMEOUT
        self._queue: asyncio.Queue[pyuavcan.transport.TransferFrom] = asyncio.Queue() 

def __init__(self,
                 specifier:        pyuavcan.transport.InputSessionSpecifier,
                 payload_metadata: pyuavcan.transport.PayloadMetadata,
                 loop:             asyncio.AbstractEventLoop,
                 finalizer:        typing.Callable[[], None]):
        """
        Do not call this directly.
        Instead, use the factory method :meth:`pyuavcan.transport.serial.SerialTransport.get_input_session`.
        """
        self._specifier = specifier
        self._payload_metadata = payload_metadata
        self._loop = loop
        assert self._loop is not None

        if not isinstance(self._specifier, pyuavcan.transport.InputSessionSpecifier) or \
                not isinstance(self._payload_metadata, pyuavcan.transport.PayloadMetadata):  # pragma: no cover
            raise TypeError('Invalid parameters')

        self._statistics = SerialInputSessionStatistics()
        self._transfer_id_timeout = self.DEFAULT_TRANSFER_ID_TIMEOUT
        self._queue: asyncio.Queue[pyuavcan.transport.TransferFrom] = asyncio.Queue()
        self._reassemblers: typing.Dict[int, TransferReassembler] = {}

        super(SerialInputSession, self).__init__(finalizer) 

def __init__(self,
                 apns_topic: str,
                 loop: Optional[asyncio.AbstractEventLoop] = None,
                 on_connection_lost: Optional[
                     Callable[['APNsBaseClientProtocol'], NoReturn]] = None,
                 auth_provider: Optional[AuthorizationHeaderProvider] = None):
        super(APNsBaseClientProtocol, self).__init__()
        self.apns_topic = apns_topic
        self.loop = loop or asyncio.get_event_loop()
        self.on_connection_lost = on_connection_lost
        self.auth_provider = auth_provider

        self.requests = {}
        self.request_streams = {}
        self.request_statuses = {}
        self.inactivity_timer = None 

def __init__(self,
                 topic: Optional[str] = None,
                 max_connections: int = 10,
                 max_connection_attempts: Optional[int] = None,
                 loop: Optional[asyncio.AbstractEventLoop] = None,
                 use_sandbox: bool = False):

        self.apns_topic = topic
        self.max_connections = max_connections
        if use_sandbox:
            self.protocol_class = APNsDevelopmentClientProtocol
        else:
            self.protocol_class = APNsProductionClientProtocol

        self.loop = loop or asyncio.get_event_loop()
        self.connections = []
        self._lock = asyncio.Lock(loop=self.loop)
        self.max_connection_attempts = max_connection_attempts 

def __init__(self,
                 key_file: str,
                 key_id: str,
                 team_id: str,
                 topic: str,
                 max_connections: int = 10,
                 max_connection_attempts: Optional[int] = None,
                 loop: Optional[asyncio.AbstractEventLoop] = None,
                 use_sandbox: bool = False):

        super(APNsKeyConnectionPool, self).__init__(
            topic=topic,
            max_connections=max_connections,
            max_connection_attempts=max_connection_attempts,
            loop=loop,
            use_sandbox=use_sandbox,
        )

        self.key_id = key_id
        self.team_id = team_id

        with open(key_file) as f:
            self.key = f.read() 

def test_app_loop_running(app):
    @app.get("/test")
    async def handler(request):
        assert isinstance(app.loop, asyncio.AbstractEventLoop)
        return text("pass")

    request, response = app.test_client.get("/test")
    assert response.text == "pass" 

def listen(self, *, loop: asyncio.AbstractEventLoop, config):
        self._loop = loop
        self._config = config
        if self._config['debug']:
            self.shutdown_grace_period = 0
            self.shutdown_wait_period = 0
            self._debug = True 

def _get_mock_loop() -> unittest.mock.Mock:
    """Return a mocked asyncio.AbstractEventLoop."""
    loop = unittest.mock.create_autospec(spec=AbstractEventLoop, spec_set=True)

    # Since calling `create_task` on our MockBot does not actually schedule the coroutine object
    # as a task in the asyncio loop, this `side_effect` calls `close()` on the coroutine object
    # to prevent "has not been awaited"-warnings.
    loop.create_task.side_effect = lambda coroutine: coroutine.close()

    return loop 

def __init__(
        self,
        loop: asyncio.AbstractEventLoop,
        host: str = 'localhost',
        port: int = 8125,
        prefix: str = None
    ):
        """Create a new client."""
        family, _, _, _, addr = socket.getaddrinfo(
            host, port, socket.AF_INET, socket.SOCK_DGRAM)[0]
        self._addr = addr
        self._prefix = prefix
        self._loop = loop
        self._transport = None 

def watch(self, loop: asyncio.AbstractEventLoop = None, callback_func: Optional[Callable] = None) -> Any:
        _loop = asyncio.get_event_loop() if not loop else loop  # type: Any

        async def _watch_loop() -> None:
            loop_counter = 0
            while True:
                loop_counter = (loop_counter + 1) % 20
                updated_files = self.update_watched_files(reindex=(loop_counter == 0))
                if updated_files:
                    added = updated_files.get('added')
                    removed = updated_files.get('removed')
                    updated = updated_files.get('updated')
                    if removed:
                        if len(removed) > 2:
                            removed[2] = '...'
                        logging.getLogger('watcher.files').warning('Removed files: {}'.format(', '.join([file for file in removed][0:3])))
                    if added:
                        if len(added) > 2:
                            added[2] = '...'
                        logging.getLogger('watcher.files').warning('New files: {}'.format(', '.join([file for file in added][0:3])))
                    if updated:
                        if len(updated) > 2:
                            updated[2] = '...'
                        logging.getLogger('watcher.files').warning('Updated files: {}'.format(', '.join([file for file in updated][0:3])))

                    if callback_func:
                        await callback_func(set([file for file in added] + [file for file in updated]))
                await asyncio.sleep(0.5)

        return _loop.create_task(_watch_loop()) 

def __init__(self, management_api: web.Application, config: Config,
                 loop: asyncio.AbstractEventLoop) -> None:
        self.loop = loop or asyncio.get_event_loop()
        self.app = web.Application(loop=self.loop, client_max_size=100 * 1024 * 1024)
        self.config = config

        self.setup_appservice()
        self.app.add_subapp(config["server.base_path"], management_api)
        self.setup_instance_subapps()
        self.setup_management_ui()

        self.runner = web.AppRunner(self.app, access_log_class=AccessLogger) 

def init(loop: asyncio.AbstractEventLoop) -> Iterable[Client]:
    Client.http_client = ClientSession(loop=loop)
    Client.loop = loop
    return Client.all() 

def init(cfg: Config, loop: AbstractEventLoop) -> web.Application:
    set_config(cfg)
    set_loop(loop)
    for pkg, enabled in cfg["api_features"].items():
        if enabled:
            importlib.import_module(f"maubot.management.api.{pkg}")
    app = web.Application(loop=loop, middlewares=[auth, error], client_max_size=100 * 1024 * 1024)
    app.add_routes(routes)
    return app 

def init(loop: asyncio.AbstractEventLoop) -> None:
    log_root.addHandler(handler)
    handler.loop = loop 

def set_loop(loop: asyncio.AbstractEventLoop) -> None:
    global _loop
    _loop = loop 

def __init__(self, client: 'MaubotMatrixClient', loop: AbstractEventLoop, http: ClientSession,
                 instance_id: str, log: Logger, config: Optional['BaseProxyConfig'],
                 database: Optional[Engine], webapp: Optional['PluginWebApp'],
                 webapp_url: Optional[str]) -> None:
        self.client = client
        self.loop = loop
        self.http = http
        self.id = instance_id
        self.log = log
        self.config = config
        self.database = database
        self.webapp = webapp
        self.webapp_url = URL(webapp_url) if webapp_url else None
        self._handlers_at_startup = [] 

def init(config: Config, webserver: 'MaubotServer', loop: AbstractEventLoop
         ) -> Iterable[PluginInstance]:
    PluginInstance.mb_config = config
    PluginInstance.loop = loop
    PluginInstance.webserver = webserver
    return PluginInstance.all() 

def server_main_logwrapper(loop: asyncio.AbstractEventLoop,
                                 pidx: int, _args: List[Any]) -> AsyncIterator[None]:
    setproctitle(f"backend.ai: manager worker-{pidx}")
    log_endpoint = _args[1]
    logger = Logger(_args[0]['logging'], is_master=False, log_endpoint=log_endpoint)
    try:
        with logger:
            async with server_main(loop, pidx, _args):
                yield
    except Exception:
        traceback.print_exc() 

def loop(*, seconds=0, minutes=0, hours=0, count=None, reconnect=True, loop=None):
    """A decorator that schedules a task in the background for you with
    optional reconnect logic. The decorator returns a :class:`Loop`.

    Parameters
    ------------
    seconds: :class:`float`
        The number of seconds between every iteration.
    minutes: :class:`float`
        The number of minutes between every iteration.
    hours: :class:`float`
        The number of hours between every iteration.
    count: Optional[:class:`int`]
        The number of loops to do, ``None`` if it should be an
        infinite loop.
    reconnect: :class:`bool`
        Whether to handle errors and restart the task
        using an exponential back-off algorithm similar to the
        one used in :meth:`discord.Client.connect`.
    loop: :class:`asyncio.AbstractEventLoop`
        The loop to use to register the task, if not given
        defaults to :func:`asyncio.get_event_loop`.

    Raises
    --------
    ValueError
        An invalid value was given.
    TypeError
        The function was not a coroutine.
    """
    def decorator(func):
        kwargs = {
            'seconds': seconds,
            'minutes': minutes,
            'hours': hours,
            'count': count,
            'reconnect': reconnect,
            'loop': loop
        }
        return Loop(func, **kwargs)
    return decorator 

def sync(coro, loop: asyncio.AbstractEventLoop):
        if asyncio.iscoroutine(coro):
            # Run async function in the loop and return the value or raise the exception
            return asyncio.run_coroutine_threadsafe(coro, loop=loop).result()

        return coro 

def __init__(self, name: str, loop: asyncio.AbstractEventLoop = None) -> None:
        self.name = name
        self._chain = []  # : List['CancelToken']
        self._triggered = asyncio.Event(loop=loop)
        self._loop = loop