Change process-local message receiving
Change process-local message receiving

The previous implementation was very racy. Suppose there are two
messages waiting in the Redis asgi message queue:
- the first receiver creates the receiver loop,
- the loop gets the first message and puts it in the appropriate queue,
- the loop is then free to loop again to get the next message,
- the receiver pops the first message off the queue
- after processing the message, the loop is cancelled;
- if it had already sent a BLPOP command, the response will be lost.

The problem can arise when using connection pools: because a
pre-existing connection won't have the initial handshake overhead, it'll
be able to fire off a BLPOP command to Redis immediately after being
taken out of the pool. The receive() code, on the other hand, won't be
aware of this and will, having processed the message it wanted, proceed
to cancel the receiver loop mid-command. If the Redis server had already
sent its response by the time the client-side connection was dropped by
the cancellation, the message will now be lost.

The root problem here is the race condition between the loop proceeding
to the next message (in cases where there aren't any other immediately
running receivers) and the final receiver cancelling it.

To avoid this, the loop has to be synchronized with its receivers.
However, merely counting the currently running receivers is
insufficient, because the loop might get messages from Redis for
channels which aren't being waited on. If this could somehow be
overcome, there is still the problem of initial startup: between
starting the loop and getting to a point where the initial receiver can
signal to the loop that it needs a message for a certain channel, the
loop might already have got to the point where it checks for its
terminating conditions and ending prematurely.

The solution implemented in this commit approaches the problem from a
different point of view. Each receiver is itself responsible for getting
messages out of Redis; if the message is for a different channel, the
old receive_buffer is used, letting any other receivers still get the
message that was just ignored. To avoid having multiple simultaneous
receivers contacting Redis independently, a token is used, which is
shared between competing receivers through an asyncio.Queue, ensuring
they all get a chance to pull from Redis. Because there is only one
token, these chances will be serialized.

As previously, the last receiver to exit also shuts down the queue; this
must happen in order to be compatible with the channel layer being used
from multiple event loops, since an asyncio.Queue will bind to an event
loop at construction time.

django/asgi_redis