checkleaks.py

from contextlib import contextmanager
import cProfile, pstats, io

import capnp
from JumpScale import j
import asyncio
import gc
import sys
import importlib
import objgraph
import random
# preload jumpscale

j.atyourservice

@contextmanager
def profile(sort_by='cumulative'):
    pr = cProfile.Profile()
    pr.enable()

    yield

    pr.disable()
    s = io.StringIO()
    sortby = 'cumulative'
    ps = pstats.Stats(pr, stream=s).sort_stats(sortby)
    ps.print_stats()
    print(s.getvalue())


def capnp_object(model, **kwargs):
    cache = []
    for i in range(1000):
        msg = model.new_message(**kwargs)
        cache.append(msg)

def python_object(model, **kwargs):
    cache = []
    for i in range(1000):
        msg = j.data.capnp.getMemoryObj(model, **kwargs)
        cache.append(msg)



def main():
    j.atyourservice._start()
    repo = j.atyourservice.aysRepos.get('/optvar/cockpit_repos/testplenty')
    for bp in repo.blueprints:
        j.atyourservice.loop.run_until_complete(repo.blueprintExecute(path=bp.path))
    run = repo.runCreate(to_execute=repo.findScheduledActions())
    j.atyourservice.loop.run_until_complete(run.execute())
    j.core.jobcontroller._methods = {}


    gc.collect()
    objgraph.show_chain(
        objgraph.find_backref_chain(
            random.choice(objgraph.by_type('JobModel')),
            objgraph.is_proper_module),
        filename='job.png')

    objgraph.show_chain(
        objgraph.find_backref_chain(
            random.choice(objgraph.by_type('JobHandler')),
            objgraph.is_proper_module),
        filename='JobHandler.png')


    run = None
    j.atyourservice.loop.close()
    # run = None
    print("collected %s" % gc.collect())
    referrers = gc.get_referrers(run)
    print("referrers %s" % len(referrers))
    print("refcount %s" % sys.getrefcount(run))
    for ref in referrers:
        print(type(ref))
        print(ref)
        print("--------------)")
    run = None
    gc.collect()
    run = None
    repo = None
    gc.collect()




if __name__ == '__main__':
    main()