Cannot cancel reading from incoming connection and retry later

[saghul]

In my application I'm doing something like this:

• Try to receive data
• Under certain conditions cancel() the task
• Try to receive data again

This yields the following assertion error:

  File "/Users/saghul/.virtualenvs/cpython342/lib/python3.4/site-packages/aiohttp/web_ws.py", line 206, in receive
    msg = yield from self._reader.read()
  File "/Users/saghul/.virtualenvs/cpython342/lib/python3.4/site-packages/aiohttp/streams.py", line 501, in read
    result = yield from super().read()
  File "/Users/saghul/.virtualenvs/cpython342/lib/python3.4/site-packages/aiohttp/streams.py", line 362, in read
    assert not self._waiter

The problem looks quite obvious, in DataQueue.read the waiter is not reset until feed_data is called: https://github.com/KeepSafe/aiohttp/blob/master/aiohttp/streams.py#L364 and https://github.com/KeepSafe/aiohttp/blob/master/aiohttp/streams.py#L343

The solution, however, is not so obvious, here is a possible solution, let me know if you like the approach and I'll submit a PR with it:

diff --git a/aiohttp/streams.py b/aiohttp/streams.py
index ce0cce4..3879ee6 100644
--- a/aiohttp/streams.py
+++ b/aiohttp/streams.py
@@ -361,7 +361,11 @@ class DataQueue:

             assert not self._waiter
             self._waiter = asyncio.Future(loop=self._loop)
-            yield from self._waiter
+            try:
+                yield from self._waiter
+            except (asyncio.CancelledError, asyncio.TimeoutError):
+                self._waiter = None
+                raise

         if self._buffer:
             data, size = self._buffer.popleft()

For now I'm setting ws._reader._waiter to None after cancelling the task , which is quite ugly... :-S

[ludovic-gasc]

For my understanding, could you describe an use case of this need? What's the interest?
Is it for a WebSockets context?

[saghul]

Here is my usecase in more detail, thanks for your interest @GMLudo!

My application is this one: https://github.com/saghul/CallRoulette

In a nutshell, if a user connects, it waits until a new user connects and when we have 2, we start processing messages between the 2.

Now, it's possible that while I wait for the second user to connect, the first one disconnects. In order to know about this, the only way is to use the receive() coroutine, AFAIK, together with my other coroutine, which waits for the second connection.

Once the 2nd connection comes, I want to cancel the receive() I started and start the processing.

In a nutshell, this looks like it was planned, there is even a related test about it, but it doesn't test the actual retrying. This also happens if you want to receive data with a timeout, AFAIS.

[kxepal]

Similar issue happens when user requests some URL that served by aiohttp with browser and hits Stop button / Esc in order to cancel page load (he tired to wait for the response, he change his mind etc.) - browser successfully terminates connection, but such assertion error occurs in logs.

Things goes wild when you have users with unstable connection (mobile ones) - your logs are full of such assertions.

[fafhrd91]

@saghul change looks good, but it hard to tell if it safe. Usually we treat cancelation as disconnection. We should just cearfuly test this change. I can test it on one of my production instances.

[lock]

This thread has been automatically locked since there has not been
any recent activity after it was closed. Please open a new issue for
related bugs.

If you feel like there's important points made in this discussion,
please include those exceprts into that new issue.