Memory leak when asyncio.open_connection raise

[seer]

BPO	44697
Nosy	@brianquinlan, @asvetlov, @1st1, @aeros, @jdevries3133

^{Note: these values reflect the state of the issue at the time it was migrated and might not reflect the current state.}

Show more details

GitHub fields:

assignee = None
closed_at = None
created_at = <Date 2021-07-21.13:46:19.757>
labels = ['expert-asyncio', '3.9', 'performance']
title = 'Memory leak when asyncio.open_connection raise'
updated_at = <Date 2021-08-05.20:05:54.604>
user = 'https://bugs.python.org/seer'

bugs.python.org fields:

activity = <Date 2021-08-05.20:05:54.604>
actor = 'jack__d'
assignee = 'none'
closed = False
closed_date = None
closer = None
components = ['asyncio']
creation = <Date 2021-07-21.13:46:19.757>
creator = 'seer'
dependencies = []
files = []
hgrepos = []
issue_num = 44697
keywords = ['patch']
message_count = 4.0
messages = ['397945', '397982', '398219', '398234']
nosy_count = 7.0
nosy_names = ['bquinlan', 'asvetlov', 'yselivanov', 'aeros', 'jack__d', 'seer', 'whoKilledLora']
pr_nums = []
priority = 'normal'
resolution = None
stage = 'patch review'
status = 'open'
superseder = None
type = 'resource usage'
url = 'https://bugs.python.org/issue44697'
versions = ['Python 3.9']

Linked PRs

• gh-88863: clear ref cycles to resolve leak when asyncio.open_connection raises #95739
• [3.11] gh-88863: Clear ref cycles to resolve leak when asyncio.open_connection raises (GH-95739) #99721
• [3.10] gh-88863: Clear ref cycles to resolve leak when asyncio.open_connection raises (GH-95739) #99722

[seer]

I write some short example.

import resource
import asyncio


class B:
    def __init__(self, loop):
        self.loop = loop
        self.some_big_data = bytearray(1024 * 1024)  # 1Mb for memory bloating

    async def doStuff(self):
        if not await self.connect():
            return
        print('Stuff done')

    async def connect(self) -> bool:
        try:
            _, writer = await asyncio.open_connection('127.0.0.1', 12345, loop=self.loop)
            writer.close()
            return True
        except OSError as e:
            pass
        return False


class A:
    def __init__(self, loop):
        self.loop = loop

    async def doBStuff(self):
        b = B(self.loop)
        await b.doStuff()

    async def work(self):
        print('Working...')
        for _ in range(1000):
            await self.loop.create_task(self.doBStuff())
        print('Done.')
        print(
            'Memory usage {}kb'.format(
                resource.getrusage(
                    resource.RUSAGE_SELF).ru_maxrss))


async def amain(loop):
    a = A(loop)
    await a.work()


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(amain(loop))

100 cycles
"Memory usage 41980kb"

1000 cycles
"Memory usage 55412kb"

10000 cycles
"Memory usage 82880kb"

And so on...

Does anyone know workaround?

[whoKilledLora]

Confirmed. I have the same problem. I suspect this is related to https://bugs.python.org/issue41699.

[seer]

Checked on 3.9.6 - still leaking.

Strange stuff, but if I write

except OSError as e:
    del self

instead of

except OSError as e:
    pass

leak is disappearing.

[aeros]

Thank you Arteem, that should help indicate where the memory leak is present.

[kumaraditya303]

This seems more like a gc issue as adding gc.collect() calls after each iteration fixes the leak.

import resource
import asyncio
import gc


class B:
    def __init__(self, loop):
        self.loop = loop
        self.some_big_data = bytearray(1024 * 1024)  # 1Mb for memory bloating

    async def doStuff(self):
        if not await self.connect():
            return
        print('Stuff done')

    async def connect(self) -> bool:
        try:
            _, writer = await asyncio.open_connection('127.0.0.1', 12345)
            writer.close()
            return True
        except OSError as e:
            pass
        return False


class A:
    def __init__(self, loop):
        self.loop = loop

    async def doBStuff(self):
        b = B(self.loop)
        await b.doStuff()

    async def work(self):
        print('Working...')
        for i in range(1000):
            await self.loop.create_task(self.doBStuff())
            gc.collect()
            if i % 100 == 0:
                print('Memory usage {}kb {}'.format(resource.getrusage(
                    resource.RUSAGE_SELF).ru_maxrss, i))
        print('Memory usage {}kb '.format(resource.getrusage(
            resource.RUSAGE_SELF).ru_maxrss, ))
        print('Done.')


async def amain(loop):
    a = A(loop)
    await a.work()


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(amain(loop))

cc @pablogsal

[pablogsal]

This seems more like a gc issue as adding gc.collect() calls after each iteration fixes the leak.

That doesn't mean it is a GC issue. That means the GC is doing its work and the problem is that there are reference cycles to clean that until the GC runs they won't go away on their own.

It may be something to help here, but the GC is working as expected from this description.

[kumaraditya303]

This might be related to #81001

[kumaraditya303]

After some investigation, the issue is that the raised exception's traceback forms ref cycles are eventually cleared by gc but causes increased memory usage. The following example does not form ref cycles and hence memory usage is constant.

import resource
import asyncio


class B:
    def __init__(self, loop):
        self.loop = loop
        self.some_big_data = bytearray(1024 * 1024)  # 1Mb for memory bloating

    async def doStuff(self):
        if not await self.connect():
            return
        print('Stuff done')

    async def connect(self) -> bool:
        try:
            _, writer = await asyncio.open_connection('127.0.0.1', 12345)
            writer.close()
            return True
        except OSError as e:
            e.__traceback__ = None # Clear ref cycle manually
        return False


class A:
    def __init__(self, loop):
        self.loop = loop

    async def doBStuff(self):
        b = B(self.loop)
        await b.doStuff()

    async def work(self):
        print('Working...')
        for i in range(1000):
            await self.loop.create_task(self.doBStuff())
            if i % 100 == 0:
                print('Memory usage {}kb {}'.format(resource.getrusage(
                    resource.RUSAGE_SELF).ru_maxrss, i))
        print('Memory usage {}kb '.format(resource.getrusage(
            resource.RUSAGE_SELF).ru_maxrss, ))
        print('Done.')


async def amain(loop):
    a = A(loop)
    await a.work()


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(amain(loop))

[frostbyte134]

As kumaraditya303 said, it seems like there are some local variables (exception and future instances) which makes reference cycles in asyncio library.
I uploaded a PR which breaks those cycles explicitly.

After the fix numbers from get_traced_memory does not grow linearly with the loop count anymore.

MacBookPro cpython % ./python.exe ~/dev/leak.py --loop_count 10  
Working...
Done.
cur = 1311527, peak = 2365922 from tracemalloc.get_traced_memory()
------------------------------
MacBookPro cpython % ./python.exe ~/dev/leak.py --loop_count 100
Working...
Done.
cur = 1317149, peak = 2371520 from tracemalloc.get_traced_memory()
------------------------------
MacBookPro cpython % ./python.exe ~/dev/leak.py --loop_count 1000
Working...
Done.
cur = 1317050, peak = 2371421 from tracemalloc.get_traced_memory()

import asyncio
import argparse
import tracemalloc


class B:
    def __init__(self):
        self.some_big_data = bytearray(1024 * 1024)  # 1Mb for memory bloating

    async def doStuff(self):
        if not await self.connect():
            return
        print('Stuff done')

    async def connect(self) -> bool:
        try:
            _, writer = await asyncio.open_connection('127.0.0.1', 12345)
            writer.close()
            return True
        except OSError as e:
            pass
        return False


class A:
    def __init__(self, loop_count):
        self.loop_count = loop_count
        pass

    async def doBStuff(self):
        b = B()
        await b.doStuff()

    async def work(self):
        print('Working...')
        for _ in range(self.loop_count):
            await self.doBStuff()
        print('Done.')
        # print(
        #     'Memory usage {}mb'.format(
        #         int(resource.getrusage(
        #             resource.RUSAGE_SELF).ru_maxrss)//1000))
        current, peak = tracemalloc.get_traced_memory()
        print(f'cur = {current}, peak = {peak} from tracemalloc.get_traced_memory()')
        print("-"*30)


async def amain(loop_count):
    a = A(loop_count)
    await a.work()


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('-l', '--loop_count', type=int, nargs='?', default=10)
    args = parser.parse_args()

    tracemalloc.start()
    asyncio.run(amain(args.loop_count))

Checked in Windows 10/Mac Monterey/Centos 7.9