ressurect_coro.py

import objgraph
import asyncio

READY = object()

import weakref
import asyncio
import gc

import logging


logger = logging.getLogger(__name__)


def create_task(coro):
    task = asyncio.create_task(coro)
    weakref.finalize(task, asyncio.create_task, coro)
    return task


async def async_fn():
    print("started")
    try:
        await asyncio.get_running_loop().create_future()
    except asyncio.CancelledError:
        print("cancelled!")
        raise
    except BaseException as e:
        logger.exception("closed!")
        raise

async def amain():
    # install_handler()
    weak_task = weakref.ref(create_task(async_fn()))
    await asyncio.sleep(0.01)
    gc.collect()
    objgraph.show_chain(
        objgraph.find_backref_chain(
            weak_task(),
            objgraph.is_proper_module
        )
    )
    await asyncio.sleep(1)


async def sheild_gc():
    # install_handler()
    strong_shielded_fut = asyncio.shield(create_task(async_fn()))
    await asyncio.sleep(0)
    asyncio.current_task().cancel()
    try:
        await strong_shielded_fut
    except asyncio.CancelledError:
        pass

    gc.collect()
    await asyncio.sleep(1)


def install_handler():
    # doesn't work - when the Task._coro is deleted before the Task gets deleted
    loop = asyncio.get_running_loop()

    def handler(loop, context, _prev_handler=loop.get_exception_handler() or loop.default_exception_handler):
        if context.get("message") != "Task was destroyed but it is pending!":
            return _prev_handler(context)

        try:
            im_running = asyncio.get_running_loop() is loop
        except RuntimeError:
            im_running = False

        breakpoint()
        if im_running:
            loop.create_task(context["task"].get_coro())
        else:
            loop.call_soon_threadsafe(loop.create_task, context["task"].get_coro())

    assert handler.__closure__ is None
    loop.set_exception_handler(handler)

asyncio.tasks.Task = asyncio.tasks._PyTask
asyncio.tasks.Future = asyncio.futures._PyFuture
asyncio.run(amain())
print("======================")
asyncio.run(sheild_gc())