def block_check(loop):
    while True:
        try:
            time.sleep(1)
            future = asyncio.run_coroutine_threadsafe(asyncio.sleep(0), loop)
            blocked_for = 0
            while True:
                try:
                    future.result(1)
                    break
                except asyncio.TimeoutError:
                    blocked_for += 1
                    task = asyncio.current_task(loop)
                    buffer = io.StringIO()
                    task.print_stack(file=buffer)
                    buffer.seek(0)
                    log.warning("Event loop blocked for longer than %d seconds (%s)\n%s\n%s" % (
                        blocked_for,
                        str(task),
                        str(last_commands),
                        buffer.read()
                    ))
        except Exception:
            pass 

def __aenter__(self):
        if self._entered:
            raise RuntimeError(
                f"TaskGroup {self!r} has been already entered")
        self._entered = True

        if self._loop is None:
            self._loop = asyncio.get_running_loop()

        self._parent_task = asyncio.current_task(self._loop)
        if self._parent_task is None:
            raise RuntimeError(
                f'TaskGroup {self!r} cannot determine the parent task')
        self._patch_task(self._parent_task)

        return self 

def get_asyncio_tasks(self, _):
        current = current_task()
        tasks = []
        for task in all_tasks():
            # Only in Python 3.8+ will we have a get_name function
            name = task.get_name() if hasattr(task, 'get_name') else getattr(task, 'name', f'Task-{id(task)}')

            task_dict = {"name": name,
                         "running": task == current,
                         "stack": [str(f) for f in task.get_stack()]}

            # Add info specific to tasks owner by TaskManager
            if hasattr(task, "start_time"):
                # Only TaskManager tasks have a start_time attribute
                cls, tsk = name.split(":")
                task_dict.update({"name": tsk, "taskmanager": cls, "start_time": task.start_time})
                if task.interval:
                    task_dict["interval"] = task.interval
            tasks.append(task_dict)
        return Response({"tasks": tasks}) 

def clean(hub, signal: int = None):
    '''
    Clean up the connections
    '''
    if signal:
        log.warning(f'Got signal {signal}! Cleaning up connections')
    coros = []
    # First clean up the remote systems
    for _, r_vals in hub.heist.ROSTERS.items():
        if not r_vals.get('bootstrap'):
            for t_name, vals in hub.heist.CONS.items():
                manager = vals['manager']
                coros.append(getattr(hub, f'heist.{manager}.clean')(t_name))
                await asyncio.gather(*coros)
    # Then shut down connections
    coros = []
    for t_name, vals in hub.heist.CONS.items():
        t_type = vals['t_type']
        coros.append(getattr(hub, f'tunnel.{t_type}.destroy')(t_name))
    await asyncio.gather(*coros)
    tasks = [t for t in asyncio.all_tasks() if t is not
             asyncio.current_task()]
    for task in tasks:
        log.warning('Task remains that were not cleaned up, shutting down violently')
        task.cancel() 

def _do_enter(self) -> 'timeout':
        # Support Tornado 5- without timeout
        # Details: https://github.com/python/asyncio/issues/392
        if self._timeout is None:
            return self

        self._task = current_task(self._loop)
        if self._task is None:
            raise RuntimeError('Timeout context manager should be used '
                               'inside a task')

        if self._timeout <= 0:
            self._loop.call_soon(self._cancel_task)
            return self

        self._cancel_at = self._loop.time() + self._timeout
        self._cancel_handler = self._loop.call_at(
            self._cancel_at, self._cancel_task)
        return self 

def __setattr__(self, name, value):
        if name in ('_loop',):
            # Set normal attributes
            object.__setattr__(self, name, value)

        else:
            # Set task local attributes
            if _GTE_PY37:
                task = asyncio.current_task(loop=self._loop)
            else:
                task = asyncio.Task.current_task(loop=self._loop)
            if task is None:
                return None

            if not hasattr(task, 'context'):
                task.context = {}

            task.context[name] = value 

def __getattribute__(self, item):
        if item in ('_loop', 'clear'):
            # Return references to local objects
            return object.__getattribute__(self, item)

        if _GTE_PY37:
            task = asyncio.current_task(loop=self._loop)
        else:
            task = asyncio.Task.current_task(loop=self._loop)
        if task is None:
            return None

        if hasattr(task, 'context') and item in task.context:
            return task.context[item]

        raise AttributeError('Task context does not have attribute {0}'.format(item)) 

def task_factory(loop, coro):
    """
    Task factory function

    Fuction closely mirrors the logic inside of
    asyncio.BaseEventLoop.create_task. Then if there is a current
    task and the current task has a context then share that context
    with the new task
    """
    task = asyncio.Task(coro, loop=loop)
    if task._source_traceback:  # flake8: noqa
        del task._source_traceback[-1]  # flake8: noqa

    # Share context with new task if possible
    if _GTE_PY37:
        current_task = asyncio.current_task(loop=loop)
    else:
        current_task = asyncio.Task.current_task(loop=loop)
    if current_task is not None and hasattr(current_task, 'context'):
        setattr(task, 'context', current_task.context)

    return task 

def test_publish(
    mock_put, mock_queue, mock_sleep, mocker, mock_uuid, mock_choices, caplog
):
    with pytest.raises(RuntimeError):  # exhausted mock_uuid list
        await mayhem.publish(mock_queue)

    exp_mock_put_calls = [
        mocker.call(mayhem.PubSubMessage(message_id="1", instance_name="cattle-1234")),
        mocker.call(mayhem.PubSubMessage(message_id="2", instance_name="cattle-5678")),
        mocker.call(mayhem.PubSubMessage(message_id="3", instance_name="cattle-9876")),
    ]
    ret_tasks = [
        t for t in asyncio.all_tasks() if t is not asyncio.current_task()
    ]
    assert 3 == len(ret_tasks)
    assert 3 == len(caplog.records)
    mock_put.assert_not_called()
    await asyncio.gather(*ret_tasks)
    assert exp_mock_put_calls == mock_put.call_args_list 

def test_consume(
    mock_get, mock_queue, message, create_mock_coro, caplog
):
    mock_get.side_effect = [message, Exception("break while loop")]
    mock_handle_message, _ = create_mock_coro("mayhem.handle_message")

    with pytest.raises(Exception, match="break while loop"):
        await mayhem.consume(mock_queue)

    ret_tasks = [
        t for t in asyncio.all_tasks() if t is not asyncio.current_task()
    ]
    assert 1 == len(ret_tasks)
    assert 1 == len(caplog.records)
    mock_handle_message.assert_not_called()
    await asyncio.gather(*ret_tasks)
    mock_handle_message.assert_called_once_with(message)


# avoid `loop.close` to actually _close_ when called in main code 

def test_consume(mock_get, mock_queue, message, create_mock_coro):
    mock_get.side_effect = [message, Exception("break while loop")]
    mock_handle_message, _ = create_mock_coro("mayhem.handle_message")

    with pytest.raises(Exception, match="break while loop"):
        await mayhem.consume(mock_queue)

    ret_tasks = [
        t for t in asyncio.all_tasks() if t is not asyncio.current_task()
    ]
    # should be 1 per side effect minus the Exception (i.e. messages consumed)
    assert 1 == len(ret_tasks)
    mock_handle_message.assert_not_called()  # <-- sanity check

    # explicitly await tasks scheduled by `asyncio.create_task`
    await asyncio.gather(*ret_tasks)

    mock_handle_message.assert_called_once_with(message) 

def shutdown(loop, executor, signal=None):
    """Cleanup tasks tied to the service's shutdown."""
    if signal:
        logging.info(f"Received exit signal {signal.name}...")
    logging.info("Closing database connections")
    logging.info("Nacking outstanding messages")
    tasks = [t for t in asyncio.all_tasks() if t is not
             asyncio.current_task()]

    [task.cancel() for task in tasks]

    logging.info(f"Cancelling {len(tasks)} outstanding tasks")
    await asyncio.gather(*tasks, return_exceptions=True)

    logging.info("Shutting down executor")
    executor.shutdown(wait=False)

    logging.info(f"Flushing metrics")
    loop.stop() 

def __aenter__(self):
        if self._entered:
            raise RuntimeError(
                f"TaskGroup {self!r} has been already entered")
        self._entered = True

        if self._loop is None:
            self._loop = asyncio.get_event_loop()

        if hasattr(asyncio, 'current_task'):
            self._parent_task = asyncio.current_task(self._loop)
        else:
            self._parent_task = asyncio.Task.current_task(self._loop)

        if self._parent_task is None:
            raise RuntimeError(
                f'TaskGroup {self!r} cannot determine the parent task')
        self._patch_task(self._parent_task)

        return self 

def _result_listener(
        self,
        result_transport: TransportPool,
        timeout: float,
        rpc_message: RpcMessage,
        return_path: str,
        options: dict,
        result_queue: InternalQueue,
    ):
        try:
            logger.debug("Result listener is waiting")
            result = await asyncio.wait_for(
                result_transport.receive_result(
                    rpc_message=rpc_message, return_path=return_path, options=options
                ),
                timeout=timeout,
            )
        except asyncio.TimeoutError as e:
            logger.debug("Result listener timed out")
            await result_queue.put(e)
        else:
            logger.debug("Result listener received result, putting onto result queue")
            await result_queue.put(result)
        finally:
            self.listener_tasks.discard(asyncio.current_task()) 

def promote(self, name, user, info=None):
        '''
        Promote the currently running task.
        '''
        task = asyncio.current_task()

        synt = getattr(task, '_syn_task', None)

        if synt is not None:

            if synt.root is None:
                return synt

            synt.root.kids.pop(synt.iden)
            synt.root = None
            return synt

        return await s_task.Task.anit(self, task, name, user, info=info) 

def process_context(self, func, model):
        """
        This function is called by an `AsyncEvent` to make callbacks processed in Event._trigger cancellable.
        Using asyncio this will result in a try-catch block catching CancelledEvents.
        Args:
            func (callable): The partial of Event._trigger with all parameters already assigned
            model (object): The currently processed model

        Returns:
            bool: returns the success state of the triggered event
        """
        if self.current_context.get() is None:
            self.current_context.set(asyncio.current_task())
            try:
                return await self._process(func)
            except asyncio.CancelledError:
                return False
        return await self._process(func) 

def _task_identity() -> int:
        loop = asyncio.get_event_loop()
        if loop.is_running():
            task = asyncio.current_task()
            task_id = id(task)
            return task_id
        else:
            return 0 

def get_task_id():
    try:
        loop = asyncio.get_event_loop()
    except RuntimeError:
        return None
    return id(current_task(loop=loop)) 

def _get_current_task(loop):  # noqa
    if sys.version_info[:2] < (3, 7):  # to avoid deprecation warning
        return asyncio.Task.current_task(loop=loop)

    else:
        return asyncio.current_task(loop=loop) 

def _shutdown(signal, loop):
    app_log.info("termination query received")
    tasks = [task for task in all_tasks() if task is not current_task()]
    [task.cancel() for task in tasks]
    app_log.warning("waiting for tasks to terminate... please wait")
    await gather(*tasks)
    loop.stop()
    app_log.info("finally exiting") 

def test_all_tasks(event_loop):
    tasks = all_tasks(event_loop)

    if PY37_OR_LATER:
        current_task = asyncio.current_task(loop=event_loop)
    else:
        current_task = asyncio.Task.current_task(loop=event_loop)

    assert current_task in tasks 

def current_task(loop: Optional[asyncio.AbstractEventLoop]=None) -> asyncio.Task:  # type: ignore  # noqa  # Return type is intentionally Generic here
    if PY_37:
        return asyncio.current_task(loop=loop)  # type: ignore
    else:
        return asyncio.Task.current_task(loop=loop)  # type: ignore 

def __enter__(self) -> BaseTimerContext:
        task = current_task(loop=self._loop)

        if task is None:
            raise RuntimeError('Timeout context manager should be used '
                               'inside a task')

        if self._cancelled:
            task.cancel()
            raise asyncio.TimeoutError from None

        self._tasks.append(task)
        return self 

def __enter__(self) -> async_timeout.timeout:
        if self._timeout is not None:
            self._task = current_task(loop=self._loop)
            if self._task is None:
                raise RuntimeError(
                    'Timeout context manager should be used inside a task')
            self._cancel_handler = self._loop.call_at(
                ceil(self._loop.time() + self._timeout), self._cancel_task)
        return self 

def current_task(loop: asyncio.AbstractEventLoop) -> 'asyncio.Task[Any]':
    if PY_37:
        task = asyncio.current_task(loop=loop)  # type: ignore
    else:
        task = asyncio.Task.current_task(loop=loop)
    if task is None:
        # this should be removed, tokio must use register_task and family API
        if hasattr(loop, 'current_task'):
            task = loop.current_task()  # type: ignore

    return task 

def shutdown(signal, loop):
    await boris_says(f"I got signal {signal.name}. Shutting down.")
    tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
    [task.cancel() for task in tasks]
    await asyncio.gather(*tasks, return_exceptions=True)
    # loop.stop() 

def clear(self):
        # If were in a task, clear the context dictionary
        if _GTE_PY37:
            task = asyncio.current_task(loop=self._loop)
        else:
            task = asyncio.Task.current_task(loop=self._loop)
        if task is not None and hasattr(task, 'context'):
            task.context.clear() 

def shutdown(loop, signal=None):
    if signal:
        logging.info(f"Received exit signal {signal.name}...")
    logging.info("Closing database connections")
    logging.info("Nacking outstanding messages")
    tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]

    [task.cancel() for task in tasks]

    logging.info("Cancelling outstanding tasks")
    await asyncio.gather(*tasks, return_exceptions=True)
    logging.info(f"Flushing metrics")
    loop.stop() 

def shutdown(loop, signal=None):
    """Cleanup tasks tied to the service's shutdown."""
    if signal:
        logging.info(f"Received exit signal {signal.name}...")
    logging.info("Closing database connections")
    logging.info("Nacking outstanding messages")
    tasks = [t for t in asyncio.all_tasks() if t is not
             asyncio.current_task()]

    [task.cancel() for task in tasks]

    logging.info(f"Cancelling {len(tasks)} outstanding tasks")
    await asyncio.gather(*tasks, return_exceptions=True)
    logging.info(f"Flushing metrics")
    loop.stop() 

def shutdown(loop, executor, signal=None):
    """Cleanup tasks tied to the service's shutdown."""
    if signal:
        logging.info(f"Received exit signal {signal.name}...")
    logging.info("Closing database connections")
    logging.info("Nacking outstanding messages")
    tasks = [t for t in asyncio.all_tasks() if t is not
             asyncio.current_task()]

    [task.cancel() for task in tasks]

    logging.info(f"Cancelling {len(tasks)} outstanding tasks")
    await asyncio.gather(*tasks, return_exceptions=True)

    logging.info("Shutting down ThreadPoolExecutor")
    executor.shutdown(wait=False)

    logging.info(f"Releasing {len(executor._threads)} threads from executor")
    for thread in executor._threads:
        try:
            thread._tstate_lock.release()
        except Exception:
            pass

    logging.info(f"Flushing metrics")
    loop.stop()