Queue.cansel() does not work in the consume mode #253
Comments
|
Could you provide RabbitMQ traffic dump? Hint command: |
|
Hey there! Although my code is slightly different, I believe its the same issue: import asyncio
import aio_pika
from aiohttp import web
async def listen_to_rmq(app):
fixture_id = "1"
conn = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
ch = await conn.channel()
ex = await ch.declare_exchange(fixture_id)
q = await ch.declare_queue(name=fixture_id, exclusive=True)
await q.bind(ex)
try:
async with q.iterator() as q_iter:
async for msg in q_iter:
async with msg.process():
print(msg.body.decode())
except asyncio.CancelledError:
pass
finally:
await q.unbind(ex, fixture_id)
await q.delete()
await ex.delete()
await ch.close()
async def start_background_tasks(app):
app['rmq_listener'] = asyncio.create_task(listen_to_rmq(app))
async def cleanup_background_tasks(app):
app['rmq_listener'].cancel()
await app['rmq_listener']
if __name__ == "__main__":
app = web.Application()
app.on_startup.append(start_background_tasks)
app.on_cleanup.append(cleanup_background_tasks)
web.run_app(app) |
ghost
mentioned this issue
|
I think I found problem. I have encountered similar issue in the past. It's related to the way how aiohttp handle stopping(and asyncio.run() because aiohttp.run_app is analogous). except (GracefulExit, KeyboardInterrupt): # pragma: no cover
pass
finally:
_cancel_all_tasks(loop)
if sys.version_info >= (3, 6): # don't use PY_36 to pass mypy
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()The problem is: logging.basicConfig(level=logging.DEBUG)in above examples: 9180.18 - aiormq.connection - DEBUG - Reader task cancelled:
Traceback (most recent call last):
File "/home/lamar/.envs/testPika/lib/python3.8/site-packages/aiormq/connection.py", line 380, in __reader
weight, channel, frame = await self.__receive_frame()
File "/home/lamar/.envs/testPika/lib/python3.8/site-packages/aiormq/connection.py", line 332, in __receive_frame
frame_header = await self.reader.readexactly(1)
File "/usr/lib/python3.8/asyncio/streams.py", line 723, in readexactly
await self._wait_for_data('readexactly')
File "/usr/lib/python3.8/asyncio/streams.py", line 517, in _wait_for_data
await self._waiter
asyncio.exceptions.CancelledErrorSo await queue.cancel() waits forever.
I think this issue is related to aio-libs/aiohttp#3593 |
|
I'm not very familiar with either aio_pika or aiormq inner workings but after taking a peek in aiormq.connection.Connection.__reader() maybe something like this could work: async def __reader(all_rpc_futures_are: asyncio.Future):
cancellation_in_progress = False
try:
while True: # not self.reader.at_eof():
if cancellation_in_progress and all_rpc_futures_are.done():
break
try:
await asyncio.sleep(0.1) # Work
except asyncio.CancelledError:
print('Socket Received Cancellation')
cancellation_in_progress = True
finally:
print('Socket close')# Somewhere in connection.close() workflow
await asyncio.gather(<all rpc futures>)
all_rpc_futures_are.set_result(None)
# Or
gathering_future = asyncio.gather(<all rpc futures>)
gathering_future.add_done_callback(lambda: all_rpc_futures_are.set_result(None))Some points that might be important:
|
|
@moznuy I think #3805 fixed the issue |
|
@multun Yes, I am quite happy that aiohttp now waits for cleanup tasks before cancelling all tasks. import asyncio
import logging
import aio_pika
logging.basicConfig(format='%(relativeCreated)8.2f - %(name)20s - %(levelname)8s - %(message)s', level=logging.DEBUG)
async def on_message(msg):
async with msg.process():
logging.info("Received: %s", msg.body.decode())
await asyncio.sleep(0.1)
async def main():
fixture_id = "test_queue"
conn: aio_pika.Connection = await aio_pika.connect("amqp://guest:guest@localhost/")
ch: aio_pika.Channel = await conn.channel()
await ch.set_qos(prefetch_count=3)
q = await ch.declare_queue(name=fixture_id, durable=True)
tag = await q.consume(callback=on_message)
try:
await asyncio.sleep(1000)
finally:
logging.debug("Before cancel")
await q.cancel(tag)
logging.debug("After cancel")
await conn.close()
if __name__ == "__main__":
asyncio.run(main())And you still get hanging on await q.cancel() even without aiohttp: Because asyncio.run works similar to aiohttp run_app before #3805 fix. One small note: I've tried a bunch of examples before stopping on this one because I wasn't getting consistent results: sometimes it was 50/50, sometimes program used to stop, sometimes adding / subtracting one line: logging.debug("Before cancel") used to change the behaviour completely upside down from 0 to 100% |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Hi. I tried aio-pika with aiohttp and noticed, that
queue.cancel()hanged infinitely. Receiving and sending messages work normally.channel.close()in the consume mode hangs too. Mentioned methods works normally in the client mode.aio-pika==6.1.2
aiormq==2.7.5
pamqp==2.3.0
aiohttp==3.6.0
RabbitMQ 3.7.18
Code example:
The text was updated successfully, but these errors were encountered: