Memory leak from `get_exchange()`

themanifold commented 3 years ago

Summary

Consistent memory leak found with the get_exchange() method.

Workaround

Use get_exchange('some_exchange', ensure=False) avoids this code path, though you need to ensure that the exchange exists before hand.

Reproduction steps:

Bring up a test docker rabbitmq server, run it with docker-compose -f compose.yml

version: "3.0"
services:
    rmq:
        image: bitnami/rabbitmq:3.8.7
        ports:
            - "15672:15672"
            - "5672:5672"
        networks:
            default:
                aliases:
                    - rmq
        environment:
            - RABBITMQ_DISK_FREE_ABSOLUTE_LIMIT=50MB

Run this toy script with python -X tracemalloc=25 toy.py 2/dev/stdout> log.txt:

import aio_pika
import asyncio
import tracemalloc

def display_top(snapshot1, key_type='filename', limit=10, count=0):
    snapshot2 = tracemalloc.take_snapshot()
    top_stats = snapshot2.compare_to(snapshot1, key_type)
    for stat in top_stats[:10]:
        print(stat)

    print("\n")

    traceback_stats = snapshot2.statistics('traceback')

    if count == 99:
        for i in range(0, 11):
            print(f"Number {i}")
            stat = traceback_stats[i]

            print("%s memory blocks: %.1f KiB" % (stat.count, stat.size / 1024))
            for line in stat.traceback.format():
                print(line)
            print("\n")

async def bootstrap():
    async def get_connection():
        return await aio_pika.connect_robust(
            host="0.0.0.0",
            login="user",
            password="bitnami",
            timeout=10,
            reconnect_interval=0
        )
    connection_pool = aio_pika.pool.Pool(get_connection, max_size=1)

    async def get_channel():
        async with connection_pool.acquire() as connection:
            return await connection.channel()

    channel_pool = aio_pika.pool.Pool(get_channel, max_size=1)

    async with channel_pool.acquire() as channel:
        await channel.declare_exchange(
            'foo',
            aio_pika.ExchangeType.TOPIC,
            timeout=2
        )
    return channel_pool

async def send_msgs(channel_pool):
    async with channel_pool.acquire() as channel:
        exchange = await channel.get_exchange("foo")
        await exchange.publish(aio_pika.Message(("Channel: %r" % channel).encode()), routing_key="bar")

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    channel_pool = loop.run_until_complete(bootstrap())
    tracemalloc.start()
    snapshot = tracemalloc.take_snapshot()
    for i in range(0, 100):
        loop.run_until_complete(send_msgs(channel_pool))
        display_top(snapshot, count=i)
        print("\n")

Look at the end of log.txt:

Number 0 374 memory blocks: 610.8 KiB File "/home/user/workspace/common/venv/lib/python3.9/site-packages/aio_pika/patterns/master.py", line 9 from aio_pika.channel import Channel File "", line 1007 File "", line 986 File "", line 680 File "", line 855 File "", line 228 File "/home/user/workspace/common/venv/lib/python3.9/site-packages/aio_pika/channel.py", line 10 from .exchange import Exchange, ExchangeType File "", line 1007 File "", line 986 File "", line 680 File "", line 855 File "", line 228 File "/home/user/workspace/common/venv/lib/python3.9/site-packages/aio_pika/exchange.py", line 8 from .message import Message File "", line 1007 File "", line 986 File "", line 680 File "", line 855 File "", line 228 File "/home/user/workspace/common/venv/lib/python3.9/site-packages/aio_pika/message.py", line 9 from pprint import pformat File "", line 1007 File "", line 986 File "", line 680 File "", line 851 File "", line 983 File "", line 647

4. Also look at the increasing size of `exchange.py`, by doing `grep 'aio_pika/exchange.py:0:' log.txt`:

grep 'aio_pika/exchange.py:0:' log.txt /home/user/workspace/common/venv/lib/python3.9/site-packages/aio_pika/exchange.py:0: size=11.2 KiB (+2064 B), count=70 (+9), average=164 B /home/user/workspace/common/venv/lib/python3.9/site-packages/aio_pika/exchange.py:0: size=11.4 KiB (+2272 B), count=73 (+12), average=160 B /home/user/workspace/common/venv/lib/python3.9/site-packages/aio_pika/exchange.py:0: size=11.6 KiB (+2480 B), count=76 (+15), average=157 B /home/user/workspace/common/venv/lib/python3.9/site-packages/aio_pika/exchange.py:0: size=11.8 KiB (+2688 B), count=79 (+18), average=154 B /home/user/workspace/common/venv/lib/python3.9/site-packages/aio_pika/exchange.py:0: size=12.0 KiB (+2896 B), count=82 (+21), average=150 B ... /home/user/workspace/common/venv/lib/python3.9/site-packages/aio_pika/exchange.py:0: size=30.5 KiB (+21.3 KiB), count=355 (+294), average=88 B /home/user/workspace/common/venv/lib/python3.9/site-packages/aio_pika/exchange.py:0: size=30.7 KiB (+21.5 KiB), count=358 (+297), average=88 B /home/user/workspace/common/venv/lib/python3.9/site-packages/aio_pika/exchange.py:0: size=30.9 KiB (+21.7 KiB), count=361 (+300), average=88 B /home/user/workspace/common/venv/lib/python3.9/site-packages/aio_pika/exchange.py:0: size=31.1 KiB (+21.9 KiB), count=364 (+303), average=88 B