Asynchronous multiprocessing.Pipe reads throw when using uvloop

Bắt đầu bằng cách chạy bản tái hiện được cung cấp với Python 3.13.5 và uvloop, rồi so sánh hành vi của nó với event loop của asyncio, tập trung vào add_reader(), multiprocessing.Pipe, poll() và recv_bytes(). Được xem là hoàn tất khi các thao tác đọc Pipe bất đồng bộ không còn phát sinh BlockingIOError dưới uvloop, đồng thời vẫn giữ nguyên hành vi đọc đã được minh họa.

Summary

When using an asyncio.Event() with loop.add_reader() to asynchronously read from a multiprocessing.Pipe, the call to recv_bytes() throws:

Process reader:
Traceback (most recent call last):
  File "/home/james/.pyenv/versions/3.13.5/lib/python3.13/multiprocessing/process.py", line 313, in _bootstrap
    self.run()
    ~~~~~~~~^^
  File "/home/james/.pyenv/versions/3.13.5/lib/python3.13/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
    ~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/james/code/connection_test.py", line 25, in reader_wrapper
    uvloop.run(async_reader(read_pipe))
    ~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/james/code/.venv/lib/python3.13/site-packages/uvloop/__init__.py", line 109, in run
    return __asyncio.run(
           ~~~~~~~~~~~~~^
        wrapper(),
        ^^^^^^^^^^
    ...<2 lines>...
        **run_kwargs
        ^^^^^^^^^^^^
    )
    ^
  File "/home/james/.pyenv/versions/3.13.5/lib/python3.13/asyncio/runners.py", line 195, in run
    return runner.run(main)
           ~~~~~~~~~~^^^^^^
  File "/home/james/.pyenv/versions/3.13.5/lib/python3.13/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
  File "uvloop/loop.pyx", line 1518, in uvloop.loop.Loop.run_until_complete
  File "/home/james/code/.venv/lib/python3.13/site-packages/uvloop/__init__.py", line 61, in wrapper
    return await main
           ^^^^^^^^^^
  File "/home/james/code/connection_test.py", line 20, in async_reader
    message = read_pipe.recv_bytes()
  File "/home/james/.pyenv/versions/3.13.5/lib/python3.13/multiprocessing/connection.py", line 216, in recv_bytes
    buf = self._recv_bytes(maxlength)
  File "/home/james/.pyenv/versions/3.13.5/lib/python3.13/multiprocessing/connection.py", line 437, in _recv_bytes
    return self._recv(size)
           ~~~~~~~~~~^^^^^^
  File "/home/james/.pyenv/versions/3.13.5/lib/python3.13/multiprocessing/connection.py", line 395, in _recv
    chunk = read(handle, remaining)
BlockingIOError: [Errno 11] Resource temporarily unavailable

This does not occur when running with asyncio

Environment

Python: 3.13.5
uvloop: 0.21.0

Reproduction

import asyncio
import uvloop
from multiprocessing import Process
from multiprocessing.connection import Connection, Pipe, wait

def writer(write_pipe: Connection):
    message = b'!' * 65536
    while True:
        write_pipe.send_bytes(message)

async def async_reader(read_pipe: Connection):
    message_available = asyncio.Event()
    asyncio.get_running_loop().add_reader(
        read_pipe.fileno(), message_available.set
    )

    while True:
        await message_available.wait()
        while read_pipe.poll():
            message = read_pipe.recv_bytes()
            print('Got message')
        message_available.clear()

def reader_wrapper(read_pipe: Connection):
    uvloop.run(async_reader(read_pipe))

if __name__ == '__main__':
    read_pipe, write_pipe = Pipe(duplex=False)

    reader_process = Process(
        name="reader", target=reader_wrapper, args=(read_pipe,)
    )
    reader_process.start()

    writer_process = Process(
        name="writer", target=writer, args=(write_pipe,)
    )
    writer_process.start()

    wait([reader_process.sentinel, writer_process.sentinel])