WebSocket handler on server side issue?

[Henddher]

Hi,

While trying to add unit tests to my project, I encountered an strange issue. Maybe a legit issue outside my code? No idea.

On the server, ws_handler waits for a condition. Until that condition is met, any incoming websocket connection is meant to wait. The handler simply loops "forever" and sends each client a "wait" message (every couple of seconds).

async def test_handler(event_loop, aiohttp_client):
    queue = asyncio.Queue(maxsize=3)

    async def ws_handler(request):
        ws = aiohttp.web.WebSocketResponse()
        await ws.prepare(request)
        i = 0
        while True:
            await asyncio.sleep(2)
            await ws.send_json({"action": "wait"})
            # await asyncio.sleep(2)
            # await queue.put(("slept", i))
            i += 1

    app = aiohttp.web.Application()
    app.router.add_get("/ws", ws_handler)
    client = await aiohttp_client(app)
    async with client.ws_connect("/ws") as ws:
        assert False  # 👈 Won't trigger!
        rawmsg = await ws.receive()
        assert (msg := ws_utils.tomsg(rawmsg))
        await queue.put((">>>> ws", ws, msg))

    while not queue.empty():
      print(await queue.get())

    assert False  # To force the test to fail and get the logs

When I run this test, the whole thing hangs. No logs, no errors (AFAICT). I was expecting that the assert marked with 👈 would happen but it does not!

So, I added an asyncio.Queue(maxsize=3) for "logging" and killing the test when the queue gets full, uncommented out the 👈 assert and here is the log:

---------------------------------------------------------------- Captured stdout call -----------------------------------------------------------------
('slept', 0)
('>>>> ws', <aiohttp.client_ws.ClientWebSocketResponse object at 0x7fae6638f4f0>, namespace(action='wait'))
('slept', 1)

Why would the ('sleep', 0) be enqueued before the client socket successfully connecting?

When I removed all asyncio.sleep(...) surrounding ws.send_json(), the test hung again. That makes me thing that the server is "starving" the client (if starvation is a thing in event loops).

Any suggestions?

Linux  5.4.0-121-generic #137-Ubuntu SMP Wed Jun 15 13:33:07 UTC 2022 x86_64 x86_64 x86_64 GNU/Linux

Python 3.10.4

 name         : aiohttp                                      
 version      : 3.8.1                                        
 description  : Async http client/server framework (asyncio) 

dependencies
 - aiosignal >=1.1.2
 - async-timeout >=4.0.0a3,<5.0
 - attrs >=17.3.0
 - charset-normalizer >=2.0,<3.0
 - frozenlist >=1.1.1
 - multidict >=4.5,<7.0
 - yarl >=1.0,<2.0

required by
 - pytest-aiohttp >=3.8.1

---

In case you wonder about ws_utils.tomsg(...):

def tomsg(rawmsg: aiohttp.WSMessage) -> Optional[SimpleNamespace]:
    logger.debug(
        "rawmsg: %s\n%s",
        rawmsg,
        "\n".join(list(reversed(traceback.format_stack()))[0:2]),
    )   
    if rawmsg.type != aiohttp.WSMsgType.TEXT:
        return None
    msg = json.loads(rawmsg.data, object_hook=lambda d: SimpleNamespace(**d))
    logger.debug("msg: %s", msg)
    return cast(SimpleNamespace, msg)

[Dreamsorcerer]

When I run this test, the whole thing hangs. No logs, no errors (AFAICT). I was expecting that the assert marked with point_left would happen but it does not!

The assert is reached, but the test doesn't seem to exit while the handler is still active. No idea why that is happening though.

> pytest test.py --timeout=3
...
----------------------------------------------------------------------------------------------- Captured log teardown -----------------------------------------------------------------------------------------------
ERROR    asyncio:base_events.py:1744 Task exception was never retrieved
future: <Task finished name='Task-1' coro=<test_handler() done, defined at /home/ubuntu/desktop/test.py:6> exception=AssertionError('assert False')>
Traceback (most recent call last):
  File "/home/ubuntu/desktop/test.py", line 24, in test_handler
    assert False  # 👈 Won't trigger!
AssertionError: assert False

Why would the ('sleep', 0) be enqueued before the client socket successfully connecting?

Not sure what you mean here. Your code awaits on receiving a message before printing the >>>> ws message, which means the handler will have already sent a response first, so the socket is connected long before either of those are printed.

Any suggestions?

No idea why the test doesn't complete as soon as the exception is hit, and continues to run the handler.
But, just use pytest-timeout for debugging the test itself, as I did above.

[Henddher]

Thank you @Dreamsorcerer

That's the fundamental issue:
Is it hanging? Why?

Just curious:
Did you try uncommenting out await queue.put(("slept", i)) in the handler?

while True:
            await asyncio.sleep(2)
            await ws.send_json({"action": "wait"})
            # await asyncio.sleep(2)
            await queue.put(("slept", i))  # 👈
            i += 1

In my case, it shows that ('sleep', 0) is happening before the >>> ws. Why would the handler go to sleep before the receiver gets the message?
Shouldn't the send stop the handler coroutine and let the receiver's coroutine run and get the message?
It would seem to me that the only time the handler relinquish control is when it calls sleep. In fact, if I removed the sleep and the queue.put(), the receiver would never run (it would "starve").

Shouldn't any async call trigger a "rescheduling" of coroutines? I would guess that it would in a round robin fashion among all those coros that are ready. But I am making assumptions about how async works when in fact I don't know.

Also thanks for the pytest timeout trick!

[Dreamsorcerer]

In my case, it shows that ('sleep', 0) is happening before the >>> ws. Why would the handler go to sleep before the receiver gets the message? Shouldn't the send stop the handler coroutine and let the receiver's coroutine run and get the message? It would seem to me that the only time the handler relinquish control is when it calls sleep. In fact, if I removed the sleep and the queue.put(), the receiver would never run (it would "starve").

OK, figured out your debugging issue. Just a silly mistake in your code. :P
You've set the queue max size to 3. So, when you remove the sleep, the handler will immediately run through the loop 3 times, fill the queue and then block on the next .put(). Then after receiving 1 of those messages, the main code also calls .put() which also blocks. There is no code running which will empty the queue, so it blocks indefinitely.

So, same advice as before. Drop the queue for debugging and go back to debugging normally (with print() or whatever) using pytest-timeout.

[Henddher]

Thank you again @Dreamsorcerer!

Good point! That's a mistake on my debugging code 🤦

However, isn't that an indication of an "starvation" behavior?
How can there be "starvation" in event loops?
Why is the handler's coroutine being scheduled all the time when the the client should get a chance to call ws.receive()?

The fundamental questions remain: "Is it hanging?" and "why?"

The assertion was my first attempt to verify if the client had a chance to ws.receive() but by never triggering suggested me that it never came out from async with client.connect() .... Why would that be?????

One thing I have not tried is adding try/catch around although I would expect that an uncaught exception within the test will stop it and produce an Error result (instead of the expected Failure due to my last assert)

[Dreamsorcerer]

Thank you again @Dreamsorcerer!

Good point! That's a mistake on my debugging code facepalm

However, isn't that an indication of an "starvation" behavior? How can there be "starvation" in event loops? Why is the handler's coroutine being scheduled all the time when the the client should get a chance to call ws.receive()?

The fundamental questions remain: "Is it hanging?" and "why?"

Still no. :P
The queue.put() will only yield to the event loop if the queue is full (which is what happened in this example).
I'm not too sure of the implementation details for .send_json(), but I'd guess it only yields once the buffer is full or similar and thus is unable to complete the request instantly.
If you want to explicitly yield to the event loop on every iteration, just add an await asyncio.sleep(0) into the loop. But, in practice this is probably fine, if you have a loop like that which just pumps vast amounts of data into the websocket, the buffer will be full in a couple of milliseconds causing it to yield.

The assertion was my first attempt to verify if the client had a chance to ws.receive() but by never triggering suggested me that it never came out from async with client.connect() .... Why would that be?????

I already pointed out that it does trigger, but because the test never concludes and gets the exception propagated, you can only see that it happens by using pytest-timeout and observing the unawaited warning.

Everything here is working as expected, except for the test runner. It seems that something is causing the handler to block the whole test run, and the test coro itself is then never awaited to get its result. This is probably a bug in the pytest_plugin or how one of the fixtures are executed, but I can't see what the cause is.

[Henddher]

Here is the minimal test (hopefully clear enough)

I added strings to each assert to differentiate.

async def test_handler(event_loop, aiohttp_client):

    async def ws_handler(request):
        ws = aiohttp.web.WebSocketResponse()
        await ws.prepare(request)
        while True:
            await asyncio.sleep(2)
            await ws.send_json({"action": "wait"})

    app = aiohttp.web.Application()
    app.router.add_get("/ws", ws_handler)
    client = await aiohttp_client(app)
    try:
      async with client.ws_connect("/ws") as ws:
          assert False, "Before receive..."
          rawmsg = await ws.receive()
          assert (msg := ws_utils.tomsg(rawmsg))
    except Exception as e:
      print("############# '", e, "' ###########")
      print(traceback.format_exc())

    assert False, "At end"
 

And using pytest --timeout=N (thanks @Dreamsorcerer for the tip!).
The test "hangs" until the timeout kicks in. I.e any value of N will kill the test after N seconds elapsed.

Here's the output:

============================= test session starts ==============================
platform linux -- Python 3.10.4, pytest-7.1.2, pluggy-1.0.0
rootdir: /home/henddher/Documents/github/myprj, configfile: pyproject.toml
plugins: asyncio-0.19.0, aiohttp-1.0.4, timeout-2.1.0
asyncio: mode=auto
timeout: 20.0s
timeout method: signal
timeout func_only: False
collected 8 items / 7 deselected / 1 selected
run-last-failure: rerun previous 1 failure
stepwise: skipping 7 already passed items.

tests/test_ws_client.py F                                                [100%]

=================================== FAILURES ===================================
_________________________________ test_handler _________________________________

args = ()
kwargs = {'aiohttp_client': <function aiohttp_client.<locals>.go at 0x7f8fd3bf52d0>, 'event_loop': <_UnixSelectorEventLoop running=False closed=False debug=False>}
coro = <coroutine object test_handler at 0x7f8fd3c00200>
task = <Task pending name='Task-2' coro=<test_handler() running at /home/henddher/Documents/github/myprj/tests/test_ws_client.py:194> wait_for=<Future pending cb=[Task.task_wakeup()]>>

    @functools.wraps(func)
    def inner(*args, **kwargs):
        coro = func(*args, **kwargs)
        if not inspect.isawaitable(coro):
            pyfuncitem.warn(
                pytest.PytestWarning(
                    f"The test {pyfuncitem} is marked with '@pytest.mark.asyncio' "
                    "but it is not an async function. "
                    "Please remove asyncio marker. "
                    "If the test is not marked explicitly, "
                    "check for global markers applied via 'pytestmark'."
                )
            )
            return
        task = asyncio.ensure_future(coro, loop=_loop)
        try:
>           _loop.run_until_complete(task)

.venv/lib/python3.10/site-packages/pytest_asyncio/plugin.py:454: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../../.pyenv/versions/3.10.4/lib/python3.10/asyncio/base_events.py:633: in run_until_complete
    self.run_forever()
../../../.pyenv/versions/3.10.4/lib/python3.10/asyncio/base_events.py:600: in run_forever
    self._run_once()
../../../.pyenv/versions/3.10.4/lib/python3.10/asyncio/base_events.py:1860: in _run_once
    event_list = self._selector.select(timeout)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selectors.EpollSelector object at 0x7f8fd3bed570>, timeout = 2.0

    def select(self, timeout=None):
        if timeout is None:
            timeout = -1
        elif timeout <= 0:
            timeout = 0
        else:
            # epoll_wait() has a resolution of 1 millisecond, round away
            # from zero to wait *at least* timeout seconds.
            timeout = math.ceil(timeout * 1e3) * 1e-3
    
        # epoll_wait() expects `maxevents` to be greater than zero;
        # we want to make sure that `select()` can be called when no
        # FD is registered.
        max_ev = max(len(self._fd_to_key), 1)
    
        ready = []
        try:
>           fd_event_list = self._selector.poll(timeout, max_ev)
E           Failed: Timeout >20.0s

../../../.pyenv/versions/3.10.4/lib/python3.10/selectors.py:469: Failed
--------------------------- Captured stdout teardown ---------------------------
############# ' Before receive...
assert False ' ###########
Traceback (most recent call last):
  File "/home/henddher/Documents/github/myprj/tests/test_ws_client.py", line 195, in test_handler
    assert False, "Before receive..."
AssertionError: Before receive...
assert False

=========================== short test summary info ============================
FAILED tests/test_ws_client.py::test_handler - Failed: Timeout >20.0s
!!!!!!!! Interrupted: Test failed, continuing from this test next run. !!!!!!!!!
======================= 1 failed, 7 deselected in 20.37s =======================

From the stack trace, it seems like the timeout is aborting a selector within asyncio from within the async with client.connect(...) as if it is in fact waiting for something that never happens (until it times out).

Any suggestion?

[Dreamsorcerer]

Right! assuming that a print() within async with client... was actually called ... but that's the blocker atm, no? frowning_face

Your previous examples were printing just fine when I ran them with a timeout.

The only way I could test this atm is by adding --timeout=N to force the test to die. True?

Yes, because the runner issue will not abort the test and produce any output otherwise.

Without the timeout, the test will never finish because of the maybe runner and/or pytest-plugin issue somewhere. True?

No, the test will never finish because you've created a test which loops indefinitely or awaits on something that will never happen. That is what you need to investigate and fix.

[Henddher]

Thank you very much @Dreamsorcerer

I replaced the while True with a counter. Then using the same client, I created many connections back to back . The test completed without any timeout. It confirmed that my "ill conceived" handler seems to the main cause.

Unfortunately, as you had also predicted, each async with client.ws_connect() gets its own handler "instance" ("server") so the local counter never increases across multiple connections. I'll add a mutable data-structure to request.app["counter"] to see what happens. I am now adding one piece at the time.

Again, thank you very much for all the help!

Hopefully, someone will figure out why the plain assert False is not causing the whole test to error out.

[Henddher]

One more note:

For the back to back connections, I ended up creating nested connections:

client = aiohttp_client(...)
async with client.ws_connect(...) as ws:
   ...
   async with client.ws_connect(...) as ws:
       ...
       async with client.ws_connect(...) as ws:
           ...

Otherwise, it would hang on the first one (never getting to the second one).

client = aiohttp_client(...)
async with client.ws_connect(...) as ws:
   ... # Hung here
async with client.ws_connect(...) as ws:
   ...
async with client.ws_connect(...) as ws:
   ...

[Dreamsorcerer]

each async with client.ws_connect() gets its own handler "instance" ("server")

It's the aiohttp_client(app) call which would create a new server. That line would just create a new WS connection. Though again, if you are counting within the handler, there would be 1 handler per connection, right? So, there'd still be a counter per call.

Otherwise, it would hang on the first one (never getting to the second one).

Not sure what the contents of the call is, but I suspect you're waiting for something in the handler to complete before allowing the WS connection to close. Depending on what you're trying to test, maybe you'd want to run multiple requests in parallel?

Something like:

async def connection(client):
    async with client.ws_connect(...) as ws:
        ...

await asyncio.gather(*(connection(client) for _ in range(3)))

[Henddher]

Thank you @Dreamsorcerer for the clarifications.

asyncio.gather() is a good idea. I'll try to leverage it.

One thing I'd like to clarify is that the sample code I've been providing is the actual code I am testing with. Originally, I was testing my own server code but when I ran into issues (test hanging), I decided to write this minimalist test: that's the test above - No other hidden/simplified changes. Maybe a branch + PR for others would help.

Thanks again!!!!!