A non-blocking read on a subprocess.PIPE in Python

Я использую модуль подпроцесса для запуска подпроцесса и подключения к его выходному потоку (стандартный вывод). Я хочу иметь возможность выполнять неблокирующие чтения в стандартном выводе. Есть ли способ сделать .readline неблокирующим или проверить, есть ли данные в потоке, прежде чем я вызову.readline? Я бы хотел, чтобы это было переносимым или, по крайней мере, работало под Windows и Linux.

Вот как я это делаю на данный момент (это блокируется в .readline, если данные недоступны):

p = subprocess.Popen('myprogram.exe', stdout = subprocess.PIPE)
output_str = p.stdout.readline()

fcntl, select, asyncproc в этом случае не поможет.

Надежный способ чтения потока без блокировки независимо от операционной системы - использовать Queue.get_nowait():

import sys
from subprocess import PIPE, Popen
from threading  import Thread

try:
    from queue import Queue, Empty
except ImportError:
    from Queue import Queue, Empty  # python 2.x

ON_POSIX = 'posix' in sys.builtin_module_names

def enqueue_output(out, queue):
    for line in iter(out.readline, b''):
        queue.put(line)
    out.close()

p = Popen(['myprogram.exe'], stdout=PIPE, bufsize=1, close_fds=ON_POSIX)
q = Queue()
t = Thread(target=enqueue_output, args=(p.stdout, q))
t.daemon = True # thread dies with the program
t.start()

# ... do other things here

# read line without blocking
try:  line = q.get_nowait() # or q.get(timeout=.1)
except Empty:
    print('no output yet')
else: # got line
    # ... do something with line

У меня часто возникала похожая проблема; программам на Python, которые я часто пишу, необходимо иметь возможность выполнять некоторые основные функциональные возможности, одновременно принимая вводимые пользователем данные из командной строки (stdin). Простое перенесение функций обработки пользовательского ввода в другой поток не решает проблему, потому что readline() блокируется и не имеет таймаута. Если основная функциональность завершена и больше нет необходимости ждать дальнейшего ввода данных пользователем, я обычно хочу, чтобы моя программа завершила работу, но это невозможно, потому что readline() все еще блокируется в другом потоке, ожидающем строки. Решение, которое я нашел для этой проблемы, состоит в том, чтобы сделать stdin неблокирующим файлом с помощью модуля fcntl:

import fcntl
import os
import sys

# make stdin a non-blocking file
fd = sys.stdin.fileno()
fl = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)

# user input handling thread
while mainThreadIsRunning:
      try: input = sys.stdin.readline()
      except: continue
      handleInput(input)

На мой взгляд, это немного чище, чем использование модулей select или signal для решения этой проблемы, но опять же, это работает только в UNIX...

В Unix-подобных системах и Python 3.5+ есть os.set_blocking который делает именно то, что написано.

import os
import time
import subprocess

cmd = 'python3', '-c', 'import time; [(print(i), time.sleep(1)) for i in range(5)]'
p = subprocess.Popen(cmd, stdout=subprocess.PIPE)
os.set_blocking(p.stdout.fileno(), False)
start = time.time()
while True:
    # first iteration always produces empty byte string in non-blocking mode
    for i in range(2):    
        line = p.stdout.readline()
        print(i, line)
        time.sleep(0.5)
    if time.time() > start + 5:
        break
p.terminate()

Это выводит:

1 b''
2 b'0\n'
1 b''
2 b'1\n'
1 b''
2 b'2\n'
1 b''
2 b'3\n'
1 b''
2 b'4\n'

С os.set_blocking комментарием это:

0 b'0\n'
1 b'1\n'
0 b'2\n'
1 b'3\n'
0 b'4\n'
1 b''

Python 3.4 представляет новый предварительный API для асинхронного ввода-вывода - asyncio модуль.

Подход аналогичен twistedоснованному на ответе @Bryan Ward - определите протокол, и его методы будут вызываться, как только данные будут готовы:

#!/usr/bin/env python3
import asyncio
import os

class SubprocessProtocol(asyncio.SubprocessProtocol):
    def pipe_data_received(self, fd, data):
        if fd == 1: # got stdout data (bytes)
            print(data)

    def connection_lost(self, exc):
        loop.stop() # end loop.run_forever()

if os.name == 'nt':
    loop = asyncio.ProactorEventLoop() # for subprocess' pipes on Windows
    asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(loop.subprocess_exec(SubprocessProtocol, 
        "myprogram.exe", "arg1", "arg2"))
    loop.run_forever()
finally:
    loop.close()

Смотрите "Подпроцесс" в документации.

Существует высокоуровневый интерфейс, asyncio.create_subprocess_exec() который возвращает Process объекты, позволяющий асинхронно считывать строку с помощью StreamReader.readline() сопрограммы (с синтаксисом async/await Python 3.5+):

#!/usr/bin/env python3.5
import asyncio
import locale
import sys
from asyncio.subprocess import PIPE
from contextlib import closing

async def readline_and_kill(*args):
    # start child process
    process = await asyncio.create_subprocess_exec(*args, stdout=PIPE)

    # read line (sequence of bytes ending with b'\n') asynchronously
    async for line in process.stdout:
        print("got line:", line.decode(locale.getpreferredencoding(False)))
        break
    process.kill()
    return await process.wait() # wait for the child process to exit


if sys.platform == "win32":
    loop = asyncio.ProactorEventLoop()
    asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()

with closing(loop):
    sys.exit(loop.run_until_complete(readline_and_kill(
        "myprogram.exe", "arg1", "arg2")))

readline_and_kill() выполняет следующие задачи:

• запустите подпроцесс, перенаправьте его стандартный вывод в канал


• асинхронное чтение строки из стандартного вывода подпроцесса


• уничтожить подпроцесс


• дождитесь завершения

При необходимости каждый шаг может быть ограничен секундами ожидания.