Default max_queue blocks websocket cancellation with high traffic
#1555
Comments
|
This is a plausible bug. If I understand correctly, to reproduce, I need to:
|
|
As a mitigation, you can use See https://websockets.readthedocs.io/en/latest/project/changelog.html and https://websockets.readthedocs.io/en/latest/howto/upgrade.html for context. |
Unfortunately I'm not sure about I tried debugging it using the following script, debugpy, and vscode, but until now I had not much time and didn't managed to access the "asyncio0" thread. Do you have any recommendations how to debug it? # repro.py
import asyncio
import logging
import json
from websockets.asyncio.client import connect
logging.getLogger(__name__)
logging.basicConfig(level=logging.DEBUG)
class MyClient:
def __init__(self):
self.keep_alive = True
async def start(self):
if hasattr(self, "task") and not self.task.done():
return
self.task: asyncio.Task = asyncio.create_task(self.run())
async def stop(self):
self.keep_alive = False
if hasattr(self, "task") and not self.task.done():
await self.task
async def run(self):
event: asyncio.Event = asyncio.Event()
async with connect(
f"wss://ws.kraken.com/v2",
ping_interval=30,
# max_queue=None, # having this enabled doesn't cause problems
) as socket:
if not event.is_set():
# subscribe to some orderbook events
# also try to reduce "depth" to 10 and just take DOT/EUR
await socket.send(
json.dumps(
{
"method": "subscribe",
"params": {
"channel": "book",
"symbol": [
"BTC/USD",
"DOT/USD",
"ETH/USD",
"MATIC/USD",
"BTC/EUR",
],
"depth": 100
},
}
)
)
event.set()
while self.keep_alive:
try:
_message = await asyncio.wait_for(socket.recv(), timeout=10)
except TimeoutError:
pass
except asyncio.CancelledError:
self.keep_alive = False
else:
try:
message = json.loads(_message)
except ValueError:
pass
else:
pass
async def __aenter__(self):
await self.start()
return self
async def __aexit__(self, *args, **kwargs):
await self.stop()
async def main():
async with MyClient():
await asyncio.sleep(2)
if __name__ == "__main__":
asyncio.run(main())BTW: Thanks for the note about the legacy asyncio stuff, but I just set |
|
I opened a pull request that addresses the issue #1556. |
|
Thank you, that helps. |
|
To make the best use of your time, I recommend that you wait until I have time to review the PR before doing more work (e.g. it isn't useful to polish the code and make CI green if we decide to take a different approach to fixing the issue). |
|
I could reproduce the problem with the following scripts: SERVER import asyncio
from websockets.asyncio.server import serve
async def count(websocket):
for i in range(20):
await websocket.send(str(i))
async def main():
async with serve(count, "localhost", 8765) as server:
await server.serve_forever()
if __name__ == "__main__":
asyncio.run(main())Client import time
from websockets.sync.client import connect
def block():
ws = connect("ws://localhost:8765", close_timeout=1)
time.sleep(1)
print("closing")
ws.close()
if __name__ == "__main__":
block()In the client, |
|
I tested if the asyncio implementation also suffers from this issue. It doesn't. |
|
The correct behavior for |
|
Oops - I just fixed a related issue in the threading implementation but I missed that this one affects the asyncio implementation. Reopening. |
|
I could reproduce the problem with the script you provided above. At first sight:
|
|
EDIT - incorrect analysis removed. |
|
I also encountered this issue in the async version of client (connecting to some other server). Everything is fine when |
|
It's not discarded; it's simply not read due to backpressure. Indeed, websockets stop reading from the network when it's receiving data faster than the application can process it. By setting |
and others
I'm working on the python-kraken-sdk and encounter strange behavior, where the websocket connection doesn't get properly closed when the connection is continuously receiving messages filling up the queue (
max_queue(https://websockets.readthedocs.io/en/stable/reference/asyncio/client.html#module-websockets.asyncio.client).This happens locally (ubuntu 24.04, Python 3.11.9) and in CI (GitHub actions, windows-latest, ubuntu-latest, py3.11 and py3.12, e.g. https://github.com/btschwertfeger/python-kraken-sdk/actions/runs/12016162113 and https://github.com/btschwertfeger/python-kraken-sdk/actions/runs/12016429517)
I'm using websockets==14.1.
In the following is the log from Ubuntu (and other runs when having
max_queuenot touched):... when setting the
max_queuetoNone(not recommended by the documentation), everything is fine:I'm not that into how the websockets implementation of yours work, but it doesn't seem to respect connections with high traffic. Proper cancellation should be done in any case, regardless of the size or number of queued messages.
This can be reproduced by running the following:
One may need to run this test multiple times in order to get the situation with too much messages during cancellation. An alternative would be to add more books to the test case (e.g.
await orderbook.add_book(pairs=["BTC/USD", "DOT/USD", "ETH/USD", "MATIC/USD", "BTC/EUR"])).The text was updated successfully, but these errors were encountered: