subscription AsyncGenerator raises error during client disconnected finalization process #889
Comments
|
After switching Python to 3.7, the stack trace changed. I believe it is a different issue in 3.7. Python 3.8 and 3.9 has the exactly same problem |
|
A possible cause due to the different behavior of asyncio in 3.8: https://bugs.python.org/issue38559
|
wuyuanyi135
changed the title
|
@wuyuanyi135 I think you're hitting a timeout or something. Can you update your logging config so that we can see timestamps please? Here's my working theory using your second code snippet:
from the docs about
Try using asyncio.wait_for: @strawberry.type
class Subscription:
@strawberry.subscription
async def test(self) -> float:
q = asyncio.Queue()
q.put_nowait(1.0)
try:
while True:
val = await asyncio.wait_for(q.get(), timeout=1.0)
yield val
await asyncio.sleep(1.0)
except asyncio.TimeoutError:
print("bye")
except BaseException as e:
print(e) |
|
@ossareh Thank you! Sorry my toy code snippet caused some confusions. I just put one item in the queue and let it wait forever to reproduce this error. The actual use case where I encountered this issue is when using a rxpy pipeline to generate real-time data and feed it to the asynchronous generator via the queue. In that case, the I have updated my code for timestamp: Codeimport asyncio
import typing
from strawberry.asgi import GraphQL
from starlette.applications import Starlette
import strawberry
import logging
logging.basicConfig(
format='%(asctime)s %(levelname)-8s %(message)s',
level=logging.INFO,
datefmt='%Y-%m-%d %H:%M:%S')
@strawberry.type
class Query:
@strawberry.field
def hello(self) -> str:
return "World"
@strawberry.type
class Subscription:
@strawberry.subscription
async def test(self) -> float:
q = asyncio.Queue()
[q.put_nowait(x) for x in range(2)]
try:
while True:
logging.info("Waiting for the queue")
val = await q.get()
logging.info("Got value from queue")
yield val
await asyncio.sleep(1.0)
except GeneratorExit:
logging.info("bye")
except BaseException as e:
logging.info(e)
schema = strawberry.Schema(Query, subscription=Subscription)
graphql = GraphQL(schema)
app = Starlette(debug=True)
app.mount("/graphql", graphql)The trace with time stamps: From this trace, I basically ended the websocket connection immediately when it started the infinite wait, so it did not seem like some time out. If I let it hang for a while in Also, I tried many different ways to kill the asynchronous generator object and the async task, but I could not reproduce the same error as seen in here - the queue could always be closed without an issue. |
|
Reproducible repo: https://github.com/wuyuanyi135/strawberry-asyncio-queue-issue |
|
There is a catch in the asgi server: strawberry/strawberry/asgi/__init__.py Line 187 in f31e6c1 This line catches Exception-derived exceptions but since 3.8 (https://docs.python.org/3.8/library/asyncio-exceptions.html#asyncio.CancelledError) it is now derived from BaseException, hence the different behavior in 3.7 and 3.8 |
|
After play with this library and the upstream Client disconnect procedures
Proposed fixTwo PRs (#897 and graphql-python/graphql-core#131) should fix the issue. strawberry/strawberry/asgi/__init__.py Line 199 in f31e6c1 |
Hello, I found a very weird problem (Python 3.8):
When run the following code and open the GraphiQL interface (http://127.0.0.1/graphql), then create a query
subscription { test }and refresh, the console will give: cannot reuse already awaited coroutine and task (detailed trace back attached later)This happens when the Queue is blocked for a while without new put into the queue, i.e. if you refresh before 2.0 is yield, it may not give you this error.
Server code
GraphQL query
Similarly, if we do not return AsyncGenerator but use the function as an AsyncGenerator, the same problem is still there:"
Trace back
After enabling
PYTHONASYNCIODEBUG=1I could see the detailed traceback now.The text was updated successfully, but these errors were encountered: