classification
Title: asyncio.stream.FlowControlMixin._drain_helper may lead to a blocking behavior
Type: Stage:
Components: asyncio Versions: Python 3.8, Python 3.7
process
Status: open Resolution:
Dependencies: Superseder:
Assigned To: Nosy List: asvetlov, malinoff, wumpus, yselivanov
Priority: normal Keywords:

Created on 2017-08-01 07:41 by malinoff, last changed 2018-05-28 16:38 by yselivanov.

Messages (4)
msg299610 - (view) Author: Dmitry Malinovsky (malinoff) Date: 2017-08-01 07:41
When there is a huge amount of `writer.write` calls followed by `await writer.drain()` on a non-paused channel, and there are no other coroutine switches, `await writer.drain()` immediately returns without a switch. This is because `asyncio.stream.FlowControlMixin._drain_helper` do not `yield` or `yield from` on a non-paused stream.

Use-case: AMQP basic.publish method, for which the broker (rabbitmq) do not send any replies back. Trying to publish 4k messages results in the following warnings (PYTHONASYNCIODEBUG env variable is set):
`Executing <Handle <TaskWakeupMethWrapper object at 0x1106fde28>(<Future finis...events.py:275>) created at /Users/malinoff/Projects/ideas/amqproto/amqproto/channel.py:85> took 2.371 seconds`
2.371 seconds is the time spent on 4k `basic_publish` calls.

You can find the test itself on github: https://github.com/malinoff/amqproto/blob/master/tests/stress/test_4k_msgs.py#L11-L12

An easy fix would be to replace return (https://github.com/python/cpython/blob/master/Lib/asyncio/streams.py#L206) with yield (and but the code below under the else clause; I'm willing to prepare a pull request), but maybe I'm missing something and such behavior is intentional?
msg317706 - (view) Author: Yury Selivanov (yselivanov) * (Python committer) Date: 2018-05-25 19:41
Andrew, what are your thoughts on this one?
msg317726 - (view) Author: Dmitry Malinovsky (malinoff) Date: 2018-05-26 04:32
I've changed the implementation significantly since August 2017, futures are not involved anymore so please ignore that part. However, such drain behavior is still present - but I don't think anymore that yielding to the event loop is an easy fix.

I've tried to do so in my lib, and it showed significant slowdowns (around 4-5k publishes per second). It's not acceptable. I also found this message from Guido https://github.com/python/asyncio/issues/263#issuecomment-142702725.

What really helped is a counter that tracks send calls without waiting for replies, and a user-provided limit; when the counter reaches the limit, an explicit yield (via await asyncio.sleep(0)) is performed. This helped to achieve around 15-16k publishes per second (3-4 times faster. Here's the code:
https://github.com/malinoff/amqproto/blob/6568204b539ecf820af2da11bddcca9ce7323ac5/amqproto/adapters/asyncio_adapter.py#L53-L71

Now I'm thinking that such behavior should only be documented - so library authors can deal with it before they face this in production. But if you have other thoughts, I'd be glad to hear.
msg317886 - (view) Author: Yury Selivanov (yselivanov) * (Python committer) Date: 2018-05-28 16:38
Documenting this would be a great first step.
History
Date User Action Args
2018-05-28 16:38:25yselivanovsetmessages: + msg317886
2018-05-26 04:32:20malinoffsetmessages: + msg317726
2018-05-25 19:41:48yselivanovsetnosy: + asvetlov

messages: + msg317706
versions: + Python 3.8, - Python 3.5, Python 3.6
2018-01-11 18:43:33wumpussetnosy: + wumpus
2017-08-01 07:41:06malinoffcreate