What is the right way to recv responses?

[altonotch]

There is something I don't understand about how to get a response messages, probably due to lack of understanding how asycio works.

Currently I have following code:

async def send_message(conn, message):
    await conn.send(payload)
    return await conn.recv()

This works fine for one call, but fails for multiple concurrent calls with error:

cannot call recv while another coroutine is already waiting for the next message

I tried using asyncio.wait_for and asyncio.ensure_future but it all fails with same error. One request succeeds the rest fail.

What would be the best way to send a message and receive a response?
Any help would be highly appreciated 🙏

[aaugustin]

Let's take a simpler example. Let's say you have a TCP stream and a coroutine that reads exactly one byte from the stream.

Now, let's say you run:

asyncio.get_event_loop().run_until_complete(
    asyncio.gather(
        stream.read_one(),
        stream.read_one(),
        stream.read_one(),
    )
)

The stream receives "abc". The three coroutines can read one byte each.

What do you expect the example to return? ["a", "b", "c"]? ["c", "b"', "a"]? Any other permutation?

All these would be equally valid.

The semantics are non-deterministic. I believe this is bad.

Now, if you replace stream.read() with conn.recv(), it's exactly the same situation, and that's why websockets prohibits concurrent calls to recv().

[deepankarm]

@aaugustin
Suppose a client sends 100 concurrent requests in a format {'id$i': $i}. Server processes (runs some operations) all the individual requests, each request may take a different amount of time.

The following code throws an error - RuntimeError: cannot call recv while another coroutine is already waiting for the next message. Why can we not let the user handle the order of the recv messages?

recv_tasks = []
for i in range(100):
   await websocket.send(f'{{id{i}: {i}}}')
   recv_tasks.append(
      asyncio.create_task(websocket.recv())
   )

for response in asyncio.as_completed(recv_tasks):
    await response
   # check id of the response & process

[aaugustin]

The main reason is that it leads to non deterministic behavior without bringing any benefit in terms concurrency, as the recv() calls will be serialized anyway. Most people hitting this error realize they made an error in their logic.

This should have pretty much the same performance characteristics and is easier to understand IMO:

for i in range(100):
   await websocket.send(f'{{id{i}: {i}}}')

for i in range(100):
    response = await websocket.recv()
    # check id of the response & process

If you really want to start reading responses before you're finished sending requests, you can run these two loops in separate tasks.

[DarkArc]

This kind of breaks a fairly beneficial asyncio property for cases where you don't need the connection to be "in lock step."

If you have some code:

while True:
  input = await websocket.recv()
  result = await process_the_request_which_may_take_a_while(input)
  await websocket.send(result)

You should be able to prevent recvs from being blocked by your processing loop by just doing:

async with asyncio.TaskGroup() as tg:
  for _ in range(os.cpu_count()):
    async def _run_my_loop():
      while True:
        input = await websocket.recv()
        result = await process_the_request_which_may_take_a_while(input)
        await websocket.send(result)
    tg.create_task(_run_my_loop())

Without that, making your send and receive loop parallel is much more awkward (producer & consumer loops and code to connect input -> processing -> output is required).

You kind of need a messaging service and a producer & consumer loop for "applications at scale" anyways, but it's a shame that you can't simply have N number of loops to process up to N number of things in parallel.

I guess for anyone hitting this exact situation, here's a rough sketch of the code you need to actually bridge the gap (call it MIT license):

# This was written to be compatible with 3.6, so the typing is a bit dated.

from typing import (
  Generic,
  TypeVar
)

InputType = TypeVar('InputType')
OutputType = TypeVar('OutputType')

class ProducerConsumerQueue(Generic[InputType, OutputType]):
  def __init__(self) -> None:
    self._input_event = asyncio.Event()
    self._output_event = asyncio.Event()
    self._input_queue: List[InputType] = []
    self._output_queue: List[OutputType] = []

  def submit_work(self, input: InputType) -> None:
    self._input_queue.append(input)
    self._input_event.set()

  async def get_work(self) -> InputType:
    while True:
      await self._input_event.wait()
      if len(self._input_queue) == 0:
        continue
      if len(self._input_queue) == 1:
        self._input_event.clear()
      return self._input_queue.pop(0)

  def submit_result(self, result: OutputType) -> None:
    self._output_queue.append(result)
    self._output_event.set()

  async def get_result(self) -> OutputType:
    while True:
      await self._output_event.wait()
      if len(self._output_queue) == 0:
        continue
      if len(self._output_queue) == 1:
        self._output_event.clear()
      return self._output_queue.pop(0)

my_queue = ProducerConsumerQueue()

async def run_input():
  while True:
    input = await websocket.recv()
    my_queue.submit_work(input)

async def run_work():
  for _ in range(os.cpu_count()):
    async def _run_my_loop():
      while True:
        await my_queue.get_work()
        result = await process_the_request_which_may_take_a_while(input)
        my_queue.submit_result(result)

    tg.create_task(_run_my_loop())

async def run_output():
  while True:
    result = await my_queue.get_result()
    await websocket.send(result)

If you want back pressure, you'll need to tweak that so that you check the size of the input queue before adding more work.

[aaugustin]

Yes, using a queue is the general idea, although you could make the implementation much simpler:

• you don't need a queue for outputs, only for inputs;
• you should use an asyncio.Queue instead of an asyncio.Event and a list.

Here's what it should look like (untested; expect stupid mistakes):

queue = asyncio.Queue()

async def worker():
  while True:
    message = await queue.get()
    result = await process_the_request_which_may_take_a_while(message)
    await websocket.send(result)

async with asyncio.TaskGroup() as tg:
  for _ in range(os.cpu_count()):
    tg.create_task(worker())

for message in websocket:
    queue.put(message)

[DarkArc]

Thanks, I'd missed that there was already a asyncio.Queue library type; that certainly makes things easier (and it looks like it also has a builtin -- optional -- queue size limit, which if used, handles the back pressure side of things as well).

It's also good to know the limitation on websockets is expected only on recv side.

I opened #1656 so that others might be (slightly) more likely to benefit from this discussion.

[aaugustin]

Indeed, send can be called concurrently safely. websockets serializes concurrent calls.

• In the standard case, send yields control to the event loop only when the buffer is full. In that case, it's good to serialize so that the second call to send wait for the buffer to drain; writing earlier would be pointless; also it would break backpressure.
• In the case of a fragmented message where fragments are generated asynchronously (not what you're doing here), serializing writes is necessary because the WebSocket protocol doesn't support interleaving messages.

(More info than you wanted, I guess!)