QueueIterator: speedup via __anext_impl w/wo timeout in __init__

[gglluukk]

Hi!

During investigation of optimization RabbitMQ's reader I've faced up performance gap between aio-pika 6.6.1 and 6.7.0 (benchmarks for 10,000 reads/seconds):

aio-pika 6.8.2: 4.484
aio-pika 6.8.1: 4.374
aio-pika 6.8.0: 4.309
aio-pika 6.7.1: 4.279
aio-pika 6.7.0: 4.436 <--- gap
aio-pika 6.6.1: 4.013
aio-pika 6.6.0: 4.083
aio-pika 6.5.3: 4.098

More: https://github.com/gglluukk/rmq2psql

The performance gap arose due to the addition of timeout support in queue.iterator(), introduced in version 6.7.0:

# diff aio_pika-6.6.1/queue.py aio_pika-6.7.0/queue.py 
439c439,442
<             return await self._queue.get()
---
>             return await asyncio.wait_for(
>                 self._queue.get(),
>                 timeout=self._consume_kwargs.get('timeout')
>             )

This change caused a decrease in performance compared to version 6.6.1 and subsequent versions inherited this gap.

PR can hopefully resolve this.

Regards!

[mosquito]

Hm... that's should be handled by wait_for implementation: https://github.com/python/cpython/blob/main/Lib/asyncio/tasks.py#L494-L507

[gglluukk]

yep... and asyncio.wait_for calls:
https://github.com/python/cpython/blob/main/Lib/asyncio/timeouts.py#L145-L162 and so on which gives ~5% overhead even if we don't use timeout= argument.
my idea is to eliminate this calls' chain if we don't need it

[Darsstar]

How does #615 do in this regard?

[gglluukk]

@Darsstar my custom task was to speed up the particular code (cancellation/connection check there is done by its own logic). during that i've researched the speed degradation and proposed patch. decision will it be accepted or not is out of my responsibility.
btw i've benchmarked your patch and it turns to be ~10% slower:

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

[coveralls]

coverage: 87.694% (-0.4%) from 88.125%
when pulling 4a89ff0 on gglluukk:master
into 848c025 on mosquito:master.

[Darsstar]

9.5.0 now contains a refactored version of the branch you benchmarked. I have incorperated the same optimizations you did, just differently. And an extra one as well. (queue.get_nowait())
Please give it a spin!