Fix handling of multiple async readers

[nomis]

If there are multiple async tasks trying to read from the WebSocket, messages spread across multiple chunks won't be handled correctly because they each independently try to combine chunks.

Only one of the multiple tasks can be successful, leaving the others starved of responses because they'll all try to read one message and won't react if a new message has already been received for them by another task.

Use a lock to ensure that only one of the tasks can be reading at any one time, and wake up the other tasks whenever there's a new message. Closed WebSockets are handled by having each task becoming the active reader in turn to discover that the connection has been closed.

[nomis]

There's a much more serious problem here in that the async WebSocket library being used don't support concurrent send and receive.

Socket reads time out after 30 seconds and I'm only listening for events which means that if there isn't an event every 30 seconds the connection is closed with a urllib3.exceptions.ReadTimeoutError exception.

If I try to send pings to keep the connection active this doesn't work because my ping is stuck waiting for a lock in self._ws.send_payload() that is held by self._ws.next_payload().

If there are multiple tasks listening for events and one of them tries to send a request because of an event, it will be blocked until the next message is received because one of the other tasks will be trying to receive its next message.

[adamlogan73]

Can you provide a minimal script that presents this behavior please? Would like to do some validation myself.

[nomis]

This script subscribes to area events that happen rarely. It will throw an error when the socket read times out after 30 seconds.

Bug 1: Pinging every second should avoid socket reads timing out but the send for the ping is blocked waiting on the recv for the event listener. While investigating the code for receiving messages, I found the following two bugs. This is an upstream issue in the async websocket library because it can't do send and receive concurrently (content warning: description contains AI slop, jawah/urllib3.future#400).

---

If there's more than one async task trying to read (which you can do by having multiple event listeners) then all of them will call recv() which observes no responses and calls _async_recv().

Inside _async_recv() there are potentially multiple calls to self._ws.next_payload() to combine fragments (although I think the library now does this for you).

Bug 2: There are multiple tasks all calling _async_recv() and it's possible for different tasks to be scheduled to run whenever await is used, so task 1 could read the first fragment and then task 2 could read the second fragment.

Bug 3: There are multiple tasks all calling recv(), observing no response and then trying to read a message and it's possible for different tasks to be scheduled to run whenever await is used, so task 2 could read the response message intended for task 1, leaving task 1 to unconditionally read another response before it can process its own response. If these are event listeners it could take a long time for another response to wake up task 1. The more tasks trying to listen for events the worse this gets. I'm reading the area, device and entity registry on startup while listening for events for changes to all of these and without my fix it can take half a minute or more to do this because it's getting delayed until my state_changed listener produces enough messages for all the relevant readers to be unblocked.

1. recv()                       2. recv()
     no response message
     call _async_recv()
                                     no response message
                                     call _async_recv()
                                       receive a message for task 1
                                     no response message
                                     call _async_recv()
       <starved of response
        waiting for another
        message>

[adamlogan73]

I am hesitant to make modifications to this client due to a deficiency in an upstream library. I see there is some discussion there, can we drive there first to see if we can fix the source? If we don't get traction there, I am willing to look at options here.

[nomis]

I am hesitant to make modifications to this client due to a deficiency in an upstream library. I see there is some discussion there, can we drive there first to see if we can fix the source? If we don't get traction there, I am willing to look at options here.

There's a bug in this library's handling of reads and upstream's locking of reads and writes.