1226.py

import asyncio
import gc
import logging
import tracemalloc
from collections import OrderedDict

from fastapi import FastAPI
from memory_profiler import memory_usage

FORMAT = '%(asctime)-15s %(message)s'
logging.basicConfig(format=FORMAT)

logger = logging.getLogger("test.mem")
logger.setLevel(logging.INFO)

app = FastAPI(
    title="MemTest",
    description="MemTest",
    version="0.0.1",
)


def show_coro(c):
    data = OrderedDict([
        ('txt', str(c)),
        ('type', str(type(c))),
        ('done', c.done()),
        ('cancelled', False),
        ('stack', None),
        ('exception', None),
    ])
    if not c.done():
        data['stack'] = [format_frame(x) for x in c.get_stack()]
    else:
        if c.cancelled():
            data['cancelled'] = True
        else:
            data['exception'] = str(c.exception())
    return data


def format_frame(f):
    keys = ['f_code', 'f_lineno']
    return OrderedDict([(k, str(getattr(f, k))) for k in keys])


async def print_mem():
    tracemalloc.start()
    while True:
        await asyncio.sleep(1)
        mem_usage = memory_usage(-1, interval=0.2, timeout=1)
        print(mem_usage)
        print(f"Number of unreachable objects {gc.collect()}")

        if tracemalloc.is_tracing():
            snapshot = tracemalloc.take_snapshot()
            top_stats = snapshot.statistics('lineno')
            for x in top_stats[:10]:
                print(str(x))

        print("+++++++++++++++++++++++++++++++++++++++++++++++++++")
        for c in asyncio.all_tasks():
            print(show_coro(c))


@app.on_event("startup")
async def on_startup():
    asyncio.gather(print_mem())


@app.get("/")
async def index():
    return 1