Как создать Asyncio Stream Reader / Writer для стандартного ввода / вывода?

Question

Мне нужно написать две программы, которые будут запускаться как родительский процесс и его дочерний процесс. Родительский процесс порождает дочерний процесс, и затем они связываются через пару каналов, связанных с дочерним stdin и stdout. Общение одноранговое, поэтому мне нужно asyncio. Простой цикл чтения / ответа не подойдет.

Я написал родителю. Нет проблем, потому что asyncio предоставляет все, что мне нужно в create_subprocess_exec().

Однако я не знаю, как создать подобную потоковую программу чтения / записи для ребенка. Я не ожидал никаких проблем. потому что каналы уже созданы и файловые дескрипторы 0 и 1 готовы к использованию при запуске дочернего процесса. Соединение не должно быть открыто, процесс не должен быть запущен.

Это моя неработающая попытка:

import asyncio
import sys

_DEFAULT_LIMIT = 64 * 1024

async def connect_stdin_stdout(limit=_DEFAULT_LIMIT, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()
    reader = asyncio.StreamReader(limit=limit, loop=loop)
    protocol = asyncio.StreamReaderProtocol(reader, loop=loop)
    r_transport, _ = await loop.connect_read_pipe(lambda: protocol, sys.stdin)
    w_transport, _ = await loop.connect_write_pipe(lambda: protocol, sys.stdout)
    writer = asyncio.StreamWriter(w_transport, protocol, reader, loop)
    return reader, writer

Проблема в том, что у меня есть два транспорта, где я должен иметь один. Функция не выполняется, потому что она пытается установить транспорт протокола дважды:

await loop.connect_read_pipe(lambda: protocol, sys.stdin)
await loop.connect_write_pipe(lambda: protocol, sys.stdout)
# !!!! assert self._transport is None, 'Transport already set'

Я пытался передать фиктивный протокол на первую строку, но эта строка также не верна, потому что нужны оба транспорта, а не один:

writer = asyncio.StreamWriter(w_transport, protocol, reader, loop)

Полагаю, мне нужно как-то объединить два однонаправленных транспорта в один двунаправленный. Или мой подход совершенно неверный? Не могли бы вы дать мне совет?

---

ОБНОВЛЕНИЕ : после некоторого теста это, кажется, работает (но мне не очень хорошо):

async def connect_stdin_stdout(limit=_DEFAULT_LIMIT, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()
    reader = asyncio.StreamReader(limit=limit, loop=loop)
    protocol = asyncio.StreamReaderProtocol(reader, loop=loop)
    dummy = asyncio.Protocol()
    await loop.connect_read_pipe(lambda: protocol, sys.stdin) # sets read_transport
    w_transport, _ = await loop.connect_write_pipe(lambda: dummy, sys.stdout)
    writer = asyncio.StreamWriter(w_transport, protocol, reader, loop)
return reader, writer

Answer 1

Ваша первая версия терпит неудачу, потому что вы используете неправильный протокол для стороны писателя; StreamReaderProtocol реализует перехватчики для реагирования на входящие соединения и данные, что сторона записи не имеет и не должна иметь дело.

Сопрограмма loop.connect_write_pipe() использует фабрику протоколов, которую вы передаете, и возвращает полученный экземпляр протокола. Вы действительно хотите использовать тот же объект протокола в модуле записи потоков вместо протокола, используемого для устройства чтения.

Далее вы не хотите передать считыватель stdin потоковому писателю stdout! Этот класс предполагает, что читатель и писатель подключены к одному и тому же файловому дескриптору, и здесь это действительно не так.

В недавнем прошлом я собрал следующее для обработки stdio для дочернего процесса; функция stdio() основана на сущности Натана Хоада по теме , а также запасной вариант для Windows, где поддержка обработки stdio как каналов ограничена .

Вы хотите, чтобы писатель правильно обрабатывал противодавление, поэтому моя версия использует (недокументированный) класс asyncio.streams.FlowControlMixin в качестве протокола для этого; вам действительно ничего больше не нужно:

import asyncio
import os
import sys

async def stdio(limit=asyncio.streams._DEFAULT_LIMIT, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()

    if sys.platform == 'win32':
        return _win32_stdio(loop)

    reader = asyncio.StreamReader(limit=limit, loop=loop)
    await loop.connect_read_pipe(
        lambda: asyncio.StreamReaderProtocol(reader, loop=loop), sys.stdin)

    writer_transport, writer_protocol = await loop.connect_write_pipe(
        lambda: asyncio.streams.FlowControlMixin(loop=loop),
        os.fdopen(sys.stdout.fileno(), 'wb'))
    writer = asyncio.streams.StreamWriter(
        writer_transport, writer_protocol, None, loop)

    return reader, writer

def _win32_stdio(loop):
    # no support for asyncio stdio yet on Windows, see https://bugs.python.org/issue26832
    # use an executor to read from stdio and write to stdout
    # note: if nothing ever drains the writer explicitly, no flushing ever takes place!
    class Win32StdinReader:
        def __init__(self):
            self.stdin = sys.stdin.buffer 
        async def readline():
            # a single call to sys.stdin.readline() is thread-safe
            return await loop.run_in_executor(None, self.stdin.readline)

    class Win32StdoutWriter:
        def __init__(self):
            self.buffer = []
            self.stdout = sys.stdout.buffer
        def write(self, data):
            self.buffer.append(data)
        async def drain(self):
            data, self.buffer = self.buffer, []
            # a single call to sys.stdout.writelines() is thread-safe
            return await loop.run_in_executor(None, sys.stdout.writelines, data)

    return Win32StdinReader(), Win32StdoutWriter()

Хотя это, возможно, немного устарело, я нашел эту запись в блоге Натаниэля Смита 2016 года об асинцио и курьезе , которая чрезвычайно полезна для понимания того, как асинхронность, протоколы, транспорт и противодавление и тому подобное все взаимодействуют и держатся вместе. В этой статье также показано, почему создание объектов чтения и записи для stdio настолько многословно и громоздко в данный момент.