QueueIterator raises StopAsyncIteration when iterator/channel is closed.

[Darsstar]

See #358

Currently QueueIterator never throws a StopAsyncIterator exception. Not when the channel is closed, and not after QueueIterator's close method is called.
Which implies that starting an async for message in queue.iterator(): loop will keep running "forever" even if no new message will ever arrive. ("forever" because the asyncio task can be canceled, etc.)

This PR fixes that in a backwards compatible way. Some tests are refactored to rely on this implicitly, and new ones that explicitly test QueueIterator.anext() throws certain exceptions have been added as well.

[coveralls]

coverage: 91.886% (+3.6%) from 88.327%
when pulling 1acd87e on Darsstar:QueueIterator-raises-StopAsyncIterator
into eb5990e on mosquito:master.

[Darsstar]

@mosquito ping

[mosquito]

@Darsstar this request contains a lot of changes that could potentially breaks backward compatibility. I'm still thinking about how to test it so that I understand I need to make a separate major release, or make do with a minor one.

[Darsstar]

this request contains a lot of changes that could potentially breaks backward compatibility.

I assume you are refering to:

• new abstract methods
• some 'protected' properties changing type (bool ->> asyncio.Event)

9.4.0 already dropped Python 3.7. The changes taking advantage of 3.8, all contained in a single commit, were made in a way it should be backward compatible.

It probably should be a new major version due to the protected properties.

Although, I think those could be rewritten as @propertys with setters which notify a asyncio.Condition ...
Than I think it boils down to wether you are supporting people inheriting from the abstract base classes without inheriting from the concrete classes aio-pika provides.
If you do the major version should be bumped.

[Darsstar]

I rebased, which added the 3.12 tests, and now tests/test_amqp_robust_proxy.py::test_channel_reconnect_stairway[[0] 0.1-128] (consistantly) fails only on 3.12, time for me to debug the issue...

[mosquito]

Thank you for not giving up on this improvement, I'm sorry that I don't have much time to devote to it right now. But I'm open to discussion.

[Darsstar]

Small update: I am currently under the impression this PR isn't at fault.

I branched from upstream/master and altered the reconnect_stairway test to start at 60 and go all the way up to 4096, instead of every power of 2 in the range [64, 4096]. The tests fail on all Python versions, so it seems to be some sort of race condition based on how tasks end up being scheduled...?
The larger the stair value, the less likely it is to fail.

I don't have a strong lead, but I started looking at aiomisc and aiormq as well. Hopefully I'll stumble on the cause in the next two weeks or so...

[LockedThread]

@LockedThread @Darsstar sorry, lots of changes, should planning to retest all these myself.

All good, we appreciate your diligent work in supporting this project.

[Darsstar]

Unsurprisingly fixing this requires more code and therefor increases the runtime.

@gglluukk's benchmark (thanks!) show about 10% runtime degredation:

before your patch:
rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #1: 4.905
rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #2: 4.503
rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #3: 4.809
rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #4: 4.870
rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #5: 4.741
Average: 4.766

after your patch:
rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #1: 5.481
rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #2: 5.123
rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #3: 5.397
rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #4: 5.437
rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #5: 5.128
Average: 5.313

[LockedThread]

Unsurprisingly fixing this requires more code and therefor increases the runtime.

@gglluukk's benchmark (thanks!) show about 10% runtime degredation:

before your patch:

rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #1: 4.905

rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #2: 4.503

rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #3: 4.809

rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #4: 4.870

rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #5: 4.741

Average: 4.766



after your patch:

rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #1: 5.481

rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #2: 5.123

rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #3: 5.397

rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #4: 5.437

rmq2psql.py --loop-type queue_iteration_with_timeout --max-bulks 100: #5: 5.128

Average: 5.313

It would be great to get the time down but I am personally willing to accept a 10% degradation in order to not have a memory leak.

[Darsstar]

It would be great to get the time down but I am personally willing to accept a 10% degradation in order to not have a memory leak.

I don't see optimisation potential without porting the library to C/C++/Rust. (While keeping the implementation correct.)

[LockedThread]

It would be great to get the time down but I am personally willing to accept a 10% degradation in order to not have a memory leak.

I don't see optimisation potential without porting the library to C/C++/Rust. (While keeping the implementation correct.)

Hopefully PyO3 gets better support for async/await, while maintaining interoperability with asyncio. There is a lot of active work on this right now. Once that happens, using Rust will be feasible.

In the meantime, losing 10% in performance is worthwhile to make sure I dont get OOM kills.

[mosquito]

You should do a performance test with cProfile for example. The last time I did this, it marshall in pamqp the slowest one.

[mosquito]

I guess we should just finish to this already. Will you fix the code style?

[Darsstar]

I guess we should just finish to this already. Will you fix the code style?

Done

[Darsstar]

I finally profiled it...

Python file I used to profile

import asyncio
import cProfile

from aio_pika import Message
from aio_pika.robust_connection import connect_robust

NUMBER_OF_MESSAGES = 1_000_000
# NUMBER_OF_MESSAGES = 1_000


async def main() -> None:
    profiler = cProfile.Profile(
        timer=asyncio.get_event_loop().time,
    )

    async with await connect_robust(host='localhost', port=5672, login='guest', password='guest') as conn:
        async with await conn.channel() as channel:
            exchange = channel.default_exchange
            queue = await channel.declare_queue('queue')

            pending = set()

            for i in range(NUMBER_OF_MESSAGES):
                pending.add(
                    asyncio.create_task(
                        exchange.publish(
                            message=Message(
                                body=f"message {i}".encode(encoding='utf-8'),
                            ),
                            routing_key=queue.name,
                            mandatory=True,
                        )
                    )
                )

            await asyncio.wait(pending)

            profiler.enable()

            count = 0

            async with queue.iterator() as queue_iter:
                async for message in queue_iter:
                    await message.ack()
                    count += 1
                    if count == NUMBER_OF_MESSAGES:
                        await queue_iter.close()
                        break

            profiler.disable()
            profiler.print_stats()
            profiler.dump_stats('D:/PR.stats')

if __name__ == '__main__':
    asyncio.run(main())

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
# before
  1000001    2.035    0.000   15.227    0.000 D:\Projects\aio-pika\aio_pika\queue.py:505(__anext__)
#after
  1000001    3.890    0.000    9.679    0.000 D:\Projects\aio-pika\aio_pika\queue.py:552(__anext__)

Regarding tottime: there is more code in QueueIterator.__anext__ in after, so it is logical more cycles are spent in that frame.
Regarding cumtime: yay! A goog 5.5 seconds faster :)

I also made CallbackCollection inherit from Generic to track down a bug I introduced, and it led me to an existing bug.

[walkingpendulum]

gentle ping with this PR, guys. any ETA on the merge? looking forward to see this released

[Darsstar]

@mosquito think you will have time before the end of the year? (It's been a month since the last ping, and the end of the year is about a months away if I factor in holidays. No more significance than that.)

If a face to face review has your preference that seems doable. My work is one metro stop away from the Nebius offices. I don't mind paying for a drink.

[LockedThread]

@mosquito think you will have time before the end of the year? (It's been a month since the last ping, and the end of the year is about a months away if I factor in holidays. No more significance than that.)

If a face to face review has your preference that seems doable. My work is one metro stop away from the Nebius offices. I don't mind paying for a drink.

I am definitely willing to chip into the beer fund if that means this gets merged!!!

[mosquito]

Thanks guys, sorry for taking so long to respond. These changes have already been published in version 9.5.0