aio-libs/aiohttp · error · RuntimeError

Concurrent call to receive() is not allowed

Error message

Concurrent call to receive() is not allowed

What it means

Raised inside ClientWebSocketResponse.receive() the moment a second coroutine enters the method while self._waiting is still True. aiohttp's websocket reader is single-threaded by design: it drains one frame at a time from one reader future, so a concurrent receive would corrupt that state. The guard makes the violation explicit instead of silently returning frames to the wrong caller.

Solutions

1. Designate exactly one task as the reader and dispatch frames to workers via an asyncio.Queue.
2. Serialize all receive calls behind a single asyncio.Lock (only one consumer acquires it at a time).
3. Cancel or await the previous receive() before starting a new one.
4. Move close()/ping() orchestration out of any task that also calls receive().

Example fix

# before
await asyncio.gather(ws.receive(), ws.receive())
# after
recv_lock = asyncio.Lock()
async def safe_receive():
    async with recv_lock:
        return await ws.receive()

class SingleReceiver:
    def __init__(self, ws):
        self._ws = ws
        self._lock = asyncio.Lock()
    async def receive(self, timeout=None):
        async with self._lock:
            return await self._ws.receive(timeout)

reader = SingleReceiver(ws)
# all tasks call reader.receive(...), never ws.receive directly

null

try:
    msg = await ws.receive()
except RuntimeError as exc:
    if 'Concurrent call' in str(exc):
        log.warning('receive serialized externally; retrying on the queue')
        await frame_queue.put(('retry', None))
    else:
        raise

When it happens

Trigger: Two tasks both awaiting ws.receive(), ws.receive_str(), ws.receive_bytes(), or ws.receive_json() on the same ClientWebSocketResponse; a receive() that was never awaited still pending when another receive() starts; calling receive() from one task while close() or a previous receive is mid-flight in another.

Common situations: Fan-out processing where each worker tries to read the shared socket; mixing a background heartbeat task with a foreground reader; forgetting to await a receive() coroutine before starting another; reusing one ws across asyncio.gather branches.

Related errors

• 1002
• Concurrent call to receive() is not allowed
• data argument must be byte-ish (%r)
• data argument must be str (%r)
• Received message : is not WSMsgType.TEXT

AI-assisted analysis of aio-libs/aiohttp@d041d4d0fd (2026-08-11). Data as JSON: /api/errors/680362de11ce8bea. Report an issue: GitHub.

Appendix: source

Thrown at aiohttp/client_ws.py:391

    @overload
    async def receive(
        self: "ClientWebSocketResponse[Literal[False]]", timeout: float | None = None
    ) -> WSMessageNoDecodeText: ...

    @overload
    async def receive(
        self: "ClientWebSocketResponse[_DecodeText]", timeout: float | None = None
    ) -> WSMessageDecodeText | WSMessageNoDecodeText: ...

    async def receive(
        self, timeout: float | None = None
    ) -> WSMessageDecodeText | WSMessageNoDecodeText:
        receive_timeout = timeout or self._timeout.ws_receive

        while True:
            if self._waiting:
                raise RuntimeError("Concurrent call to receive() is not allowed")

            if self._closed:
                return WS_CLOSED_MESSAGE
            elif self._closing:
                await self.close()
                return WS_CLOSED_MESSAGE

            try:
                self._waiting = True
                try:
                    if receive_timeout:
                        # Entering the context manager and creating
                        # Timeout() object can take almost 50% of the
                        # run time in this loop so we avoid it if
                        # there is no read timeout.
                        async with async_timeout.timeout(receive_timeout):
                            msg = await self._reader.read()
                    else:

View on GitHub (pinned to d041d4d0fd)