非阻塞读取python中的subprocess.PIPE
我正在使用subprocess模块来启动一个subprocess并连接到它的输出stream（标准输出）。 我想能够在其stdout上执行非阻塞读取。 有没有办法让.readline非阻塞或在我调用.readline之前检查stream中是否有数据？ 我希望这是可移植的，或者至less在Windows和Linux下工作。
这里是我现在是怎么做的（如果没有数据可用，它会阻止在.readline ）：
p = subprocess.Popen('myprogram.exe', stdout = subprocess.PIPE) output_str = p.stdout.readline()
fcntl ， select ， asyncproc在这种情况下asyncproc 。
无论操作系统如何，无阻塞地读取stream的可靠方法是使用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 3.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程序需要能够执行一些主要function，同时接受来自命令行（stdin）的用户input。 简单地把用户input处理function放在另一个线程中并不能解决问题，因为readline()块并没有超时。 如果主要function已经完成，并且不再需要等待进一步的用户input，我通常希望我的程序退出，但是它不能，因为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或信号模块来解决这个问题稍微清洁一些，但它只能在UNIX上工作…
Python 3.4为asynchronousIO引入了新的临时API – asyncio模块 。
该方法类似于@Bryan Ward的基于twisted的答案 – 定义一个协议，并在数据准备就绪后立即调用它的方法：
#!/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()
请参阅文档中的“子stream程” 。
有一个高级接口asyncio.create_subprocess_exec() ，返回Process对象 ，允许使用StreamReader.readline()协同程序asynchronous读取一行（使用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()执行以下任务：
- 启动subprocess，将其stdoutredirect到pipe道
- 从subprocessstdoutasynchronous读取一行
- 杀死subprocess
- 等待它退出
如有必要，每个步骤都可能受到超时秒数的限制。
尝试asyncproc模块。 例如：
import os from asyncproc import Process myProc = Process("myprogram.app") while True: # check to see if process has ended poll = myProc.wait(os.WNOHANG) if poll != None: break # print any new output out = myProc.read() if out != "": print out
该模块负责S.Lott所build议的所有线程处理。
你可以在Twisted中很容易地做到这一点。 根据你现有的代码库，这可能不是那么容易使用，但如果你正在构build一个扭曲的应用程序，那么这样的事情变得几乎微不足道。 您创build一个ProcessProtocol类，并覆盖outReceived()方法。 扭曲（取决于所使用的反应堆）通常只是一个大的select()循环，安装callback函数来处理来自不同文件描述符（通常是networking套接字）的数据。 所以outReceived()方法只是安装一个处理来自STDOUT数据的callbackoutReceived() 。 演示此行为的一个简单示例如下所示：
from twisted.internet import protocol, reactor class MyProcessProtocol(protocol.ProcessProtocol): def outReceived(self, data): print data proc = MyProcessProtocol() reactor.spawnProcess(proc, './myprogram', ['./myprogram', 'arg1', 'arg2', 'arg3']) reactor.run()
扭曲的文档有这方面的一些很好的信息。
如果你围绕Twisted构build你的整个应用程序，它会与其他本地或远程进程进行asynchronous通信，就像这样。 另一方面，如果你的程序不是在Twisted的基础上构build的，那么这不会对你有帮助。 希望这可以帮助其他读者，即使它不适用于您的特定应用程序。
使用select＆read（1）。
import subprocess #no new requirements def readAllSoFar(proc, retVal=''): while (select.select([proc.stdout],[],[],0)[0]!=[]): retVal+=proc.stdout.read(1) return retVal p = subprocess.Popen(['/bin/ls'], stdout=subprocess.PIPE) while not p.poll(): print (readAllSoFar(p))
对于readline（） – 就像：
lines = [''] while not p.poll(): lines = readAllSoFar(p, lines[-1]).split('\n') for a in range(len(lines)-1): print a lines = readAllSoFar(p, lines[-1]).split('\n') for a in range(len(lines)-1): print a
一个解决scheme是让另一个进程来执行你的进程的读取，或者使一个进程的线程超时。
这是一个超时函数的线程版本：
http://code.activestate.com/recipes/473878/
但是，你是否需要阅读stdout进来？ 另一个解决scheme可能是将输出转储到一个文件，并等待使用p.wait（）完成该过程。
f = open('myprogram_output.txt','w') p = subprocess.Popen('myprogram.exe', stdout=f) p.wait() f.close() str = open('myprogram_output.txt','r').read()
免责声明：这只适用于龙卷风
你可以通过设置fd为非阻塞，然后使用ioloop来注册callback。 我把这个包装在一个名为tornado_subprocess的蛋中，你可以通过PyPI安装它：
easy_install tornado_subprocess
现在你可以做这样的事情：
import tornado_subprocess import tornado.ioloop def print_res( status, stdout, stderr ) : print status, stdout, stderr if status == 0: print "OK:" print stdout else: print "ERROR:" print stderr t = tornado_subprocess.Subprocess( print_res, timeout=30, args=[ "cat", "/etc/passwd" ] ) t.start() tornado.ioloop.IOLoop.instance().start()
你也可以使用RequestHandler
class MyHandler(tornado.web.RequestHandler): def on_done(self, status, stdout, stderr): self.write( stdout ) self.finish() @tornado.web.asynchronous def get(self): t = tornado_subprocess.Subprocess( self.on_done, timeout=30, args=[ "cat", "/etc/passwd" ] ) t.start()
现有的解决scheme不适合我（详情如下）。 最后的工作是使用read（1）来实现readline（基于这个答案 ）。 后者不阻挡：
from subprocess import Popen, PIPE from threading import Thread def process_output(myprocess): #output-consuming thread nextline = None buf = '' while True: #--- extract line using read(1) out = myprocess.stdout.read(1) if out == '' and myprocess.poll() != None: break if out != '': buf += out if out == '\n': nextline = buf buf = '' if not nextline: continue line = nextline nextline = None #--- do whatever you want with line here print 'Line is:', line myprocess.stdout.close() myprocess = Popen('myprogram.exe', stdout=PIPE) #output-producing process p1 = Thread(target=process_output, args=(dcmpid,)) #output-consuming thread p1.daemon = True p1.start() #--- do whatever here and then kill process and thread if needed if myprocess.poll() == None: #kill process; will automatically stop thread myprocess.kill() myprocess.wait() if p1 and p1.is_alive(): #wait for thread to finish p1.join()
为什么现有解决scheme无效：
- 需要readline的解决scheme（包括基于队列的解决scheme）始终阻塞。 杀死执行readline的线程是困难的（不可能？）。 它只会在创build它的过程完成时被杀死，而不会在产生输出的过程中被杀死。
- 如同anonnn指出的那样，将低级fcntl与高级readline调用混合可能无法正常工作。
- 使用select.poll（）是整洁，但不能在Windows上根据python文档工作。
- 使用第三方库似乎矫枉过正这个任务，并增加了额外的依赖。
我添加这个问题来读取一些subprocess.Popen标准输出。 这是我的非阻塞阅读解决scheme：
import fcntl def non_block_read(output): fd = output.fileno() fl = fcntl.fcntl(fd, fcntl.F_GETFL) fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK) try: return output.read() except: return "" # Use example from subprocess import * sb = Popen("echo test && sleep 1000", shell=True, stdout=PIPE) sb.kill() # sb.stdout.read() # <-- This will block non_block_read(sb.stdout) 'test\n'
这是我的代码，用于捕获subprocessASAP的每个输出，包括部分行。 它在几乎正确的顺序同时泵和stdout和stderr。
经过testing，正确使用Python 2.7 linux＆windows。
#!/usr/bin/python # # Runner with stdout/stderr catcher # from sys import argv from subprocess import Popen, PIPE import os, io from threading import Thread import Queue def __main__(): if (len(argv) > 1) and (argv[-1] == "-sub-"): import time, sys print "Application runned!" time.sleep(2) print "Slept 2 second" time.sleep(1) print "Slept 1 additional second", time.sleep(2) sys.stderr.write("Stderr output after 5 seconds") print "Eol on stdin" sys.stderr.write("Eol on stderr\n") time.sleep(1) print "Wow, we have end of work!", else: os.environ["PYTHONUNBUFFERED"]="1" try: p = Popen( argv + ["-sub-"], bufsize=0, # line-buffered stdin=PIPE, stdout=PIPE, stderr=PIPE ) except WindowsError, W: if W.winerror==193: p = Popen( argv + ["-sub-"], shell=True, # Try to run via shell bufsize=0, # line-buffered stdin=PIPE, stdout=PIPE, stderr=PIPE ) else: raise inp = Queue.Queue() sout = io.open(p.stdout.fileno(), 'rb', closefd=False) serr = io.open(p.stderr.fileno(), 'rb', closefd=False) def Pump(stream, category): queue = Queue.Queue() def rdr(): while True: buf = stream.read1(8192) if len(buf)>0: queue.put( buf ) else: queue.put( None ) return def clct(): active = True while active: r = queue.get() try: while True: r1 = queue.get(timeout=0.005) if r1 is None: active = False break else: r += r1 except Queue.Empty: pass inp.put( (category, r) ) for tgt in [rdr, clct]: th = Thread(target=tgt) th.setDaemon(True) th.start() Pump(sout, 'stdout') Pump(serr, 'stderr') while p.poll() is None: # App still working try: chan,line = inp.get(timeout = 1.0) if chan=='stdout': print "STDOUT>>", line, "<?<" elif chan=='stderr': print " ERROR==", line, "=?=" except Queue.Empty: pass print "Finish" if __name__ == '__main__': __main__()
此版本的非阻塞式读取不需要特殊的模块，并且可以在大多数Linux发行版上直接使用。
import os import sys import time import fcntl import subprocess def async_read(fd): # set non-blocking flag while preserving old flags fl = fcntl.fcntl(fd, fcntl.F_GETFL) fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK) # read char until EOF hit while True: try: ch = os.read(fd.fileno(), 1) # EOF if not ch: break sys.stdout.write(ch) except OSError: # waiting for data be available on fd pass def shell(args, async=True): # merge stderr and stdout proc = subprocess.Popen(args, shell=False, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) if async: async_read(proc.stdout) sout, serr = proc.communicate() return (sout, serr) if __name__ == '__main__': cmd = 'ping 8.8.8.8' sout, serr = shell(cmd.split())
select模块可帮助您确定下一个有用input的位置。
但是，你几乎总是对单独的线程感到高兴。 一个阻塞读取标准input，另一个不pipe你不想阻塞的地方。
在这里添加这个答案，因为它提供了在Windows和Unix上设置非阻塞pipe道的能力。
所有的ctypes细节都归功于@ techtonik的回答 。
在Unix和Windows系统上都有一个稍微修改的版本。
- Python3兼容（只需稍作更改） 。
- 包含posix版本，并定义要使用的exception。
这样你可以对Unix和Windows代码使用相同的函数和例外。
# pipe_non_blocking.py (module) """ Example use: p = subprocess.Popen( command, stdout=subprocess.PIPE, ) pipe_non_blocking_set(p.stdout.fileno()) try: data = os.read(p.stdout.fileno(), 1) except PortableBlockingIOError as ex: if not pipe_non_blocking_is_error_blocking(ex): raise ex """ __all__ = ( "pipe_non_blocking_set", "pipe_non_blocking_is_error_blocking", "PortableBlockingIOError", ) import os if os.name == "nt": def pipe_non_blocking_set(fd): # Constant could define globally but avoid polluting the name-space # thanks to: https://stackoverflow.com/questions/34504970 import msvcrt from ctypes import windll, byref, wintypes, WinError, POINTER from ctypes.wintypes import HANDLE, DWORD, BOOL LPDWORD = POINTER(DWORD) PIPE_NOWAIT = wintypes.DWORD(0x00000001) def pipe_no_wait(pipefd): SetNamedPipeHandleState = windll.kernel32.SetNamedPipeHandleState SetNamedPipeHandleState.argtypes = [HANDLE, LPDWORD, LPDWORD, LPDWORD] SetNamedPipeHandleState.restype = BOOL h = msvcrt.get_osfhandle(pipefd) res = windll.kernel32.SetNamedPipeHandleState(h, byref(PIPE_NOWAIT), None, None) if res == 0: print(WinError()) return False return True return pipe_no_wait(fd) def pipe_non_blocking_is_error_blocking(ex): if not isinstance(ex, PortableBlockingIOError): return False from ctypes import GetLastError ERROR_NO_DATA = 232 return (GetLastError() == ERROR_NO_DATA) PortableBlockingIOError = OSError else: def pipe_non_blocking_set(fd): import fcntl fl = fcntl.fcntl(fd, fcntl.F_GETFL) fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK) return True def pipe_non_blocking_is_error_blocking(ex): if not isinstance(ex, PortableBlockingIOError): return False return True PortableBlockingIOError = BlockingIOError
为了避免读取不完整的数据，我写了自己的readline生成器（它返回每行的字节串）。
它的发电机，所以你可以例如…
def non_blocking_readlines(f, chunk=1024): """ Iterate over lines, yielding b'' when nothings left or when new data is not yet available. stdout_iter = iter(non_blocking_readlines(process.stdout)) line = next(stdout_iter) # will be a line or b''. """ import os from .pipe_non_blocking import ( pipe_non_blocking_set, pipe_non_blocking_is_error_blocking, PortableBlockingIOError, ) fd = f.fileno() pipe_non_blocking_set(fd) blocks = [] while True: try: data = os.read(fd, chunk) if not data: # case were reading finishes with no trailing newline yield b''.join(blocks) blocks.clear() except PortableBlockingIOError as ex: if not pipe_non_blocking_is_error_blocking(ex): raise ex yield b'' continue while True: n = data.find(b'\n') if n == -1: break yield b''.join(blocks) + data[:n + 1] data = data[n + 1:] blocks.clear() blocks.append(data)
我已经基于JF Sebastian的解决scheme创build了一个库。 你可以使用它。
从JF Sebastian的回答和其他一些资料来看，我已经组build了一个简单的子stream程pipe理器。 它提供了请求非阻塞读取，以及并行运行多个进程。 它不使用任何操作系统特定的调用（我知道），因此应该在任何地方工作。
它可以从pypi，所以只需pip install shelljob 。 参考项目页面的例子和完整的文档。
编辑：这个实现仍然阻止。 改用JFSebastian的回答 。
我尝试了最重要的答案 ，但线程代码的额外风险和维护令人担忧。
通过io模块 （限于2.6），我发现了BufferedReader。 这是我的无线解决scheme。
import io from subprocess import PIPE, Popen p = Popen(['myprogram.exe'], stdout=PIPE) SLEEP_DELAY = 0.001 # Create an io.BufferedReader on the file descriptor for stdout with io.open(p.stdout.fileno(), 'rb', closefd=False) as buffer: while p.poll() == None: time.sleep(SLEEP_DELAY) while '\n' in bufferedStdout.peek(bufferedStdout.buffer_size): line = buffer.readline() # do stuff with the line # Handle any remaining output after the process has ended while buffer.peek(): line = buffer.readline() # do stuff with the line
我最近偶然发现了同样的问题，我需要在非阻塞模式下从stream（在subprocess中的尾部运行）读取一行，我想避免接下来的问题：不要刻录CPU，不要读取一个字节的stream（像readline那样）等等
这里是我的实现https://gist.github.com/grubberr/5501e1a9760c3eab5e0a它不支持窗口（轮询），不处理EOF，但它对我很好;
为什么麻烦线程和队列？ 不像readline（），BufferedReader.read1（）不会阻塞等待\ r \ n，如果有任何输出进来，它会立即返回。
#!/usr/bin/python from subprocess import Popen, PIPE, STDOUT import io def __main__(): try: p = Popen( ["ping", "-n", "3", "127.0.0.1"], stdin=PIPE, stdout=PIPE, stderr=STDOUT ) except: print("Popen failed"); quit() sout = io.open(p.stdout.fileno(), 'rb', closefd=False) while True: buf = sout.read1(1024) if len(buf) == 0: break print buf, if __name__ == '__main__': __main__()
在我的情况下，我需要一个logging模块，捕捉来自后台应用程序的输出，并增加它（添加时间戳，颜色等）。
我结束了后台线程，实际的I / O。 以下代码仅适用于POSIX平台。 我剥离了非必要的部分。
如果有人打算长时间使用这个野兽考虑pipe理开放描述符。 在我的情况下，这不是一个大问题。
# -*- python -*- import fcntl import threading import sys, os, errno import subprocess class Logger(threading.Thread): def __init__(self, *modules): threading.Thread.__init__(self) try: from select import epoll, EPOLLIN self.__poll = epoll() self.__evt = EPOLLIN self.__to = -1 except: from select import poll, POLLIN print 'epoll is not available' self.__poll = poll() self.__evt = POLLIN self.__to = 100 self.__fds = {} self.daemon = True self.start() def run(self): while True: events = self.__poll.poll(self.__to) for fd, ev in events: if (ev&self.__evt) != self.__evt: continue try: self.__fds[fd].run() except Exception, e: print e def add(self, fd, log): assert not self.__fds.has_key(fd) self.__fds[fd] = log self.__poll.register(fd, self.__evt) class log: logger = Logger() def __init__(self, name): self.__name = name self.__piped = False def fileno(self): if self.__piped: return self.write self.read, self.write = os.pipe() fl = fcntl.fcntl(self.read, fcntl.F_GETFL) fcntl.fcntl(self.read, fcntl.F_SETFL, fl | os.O_NONBLOCK) self.fdRead = os.fdopen(self.read) self.logger.add(self.read, self) self.__piped = True return self.write def __run(self, line): self.chat(line, nl=False) def run(self): while True: try: line = self.fdRead.readline() except IOError, exc: if exc.errno == errno.EAGAIN: return raise self.__run(line) def chat(self, line, nl=True): if nl: nl = '\n' else: nl = '' sys.stdout.write('[%s] %s%s' % (self.__name, line, nl)) def system(command, param=[], cwd=None, env=None, input=None, output=None): args = [command] + param p = subprocess.Popen(args, cwd=cwd, stdout=output, stderr=output, stdin=input, env=env, bufsize=0) p.wait() ls = log('ls') ls.chat('go') system("ls", ['-l', '/'], output=ls) date = log('date') date.chat('go') system("date", output=date)
This is a example to run interactive command in subprocess, and the stdout is interactive by using pseudo terminal. You can refer to: https://stackoverflow.com/a/43012138/3555925
#!/usr/bin/env python # -*- coding: utf-8 -*- import os import sys import select import termios import tty import pty from subprocess import Popen command = 'bash' # command = 'docker run -it --rm centos /bin/bash'.split() # save original tty setting then set it to raw mode old_tty = termios.tcgetattr(sys.stdin) tty.setraw(sys.stdin.fileno()) # open pseudo-terminal to interact with subprocess master_fd, slave_fd = pty.openpty() # use os.setsid() make it run in a new process group, or bash job control will not be enabled p = Popen(command, preexec_fn=os.setsid, stdin=slave_fd, stdout=slave_fd, stderr=slave_fd, universal_newlines=True) while p.poll() is None: r, w, e = select.select([sys.stdin, master_fd], [], []) if sys.stdin in r: d = os.read(sys.stdin.fileno(), 10240) os.write(master_fd, d) elif master_fd in r: o = os.read(master_fd, 10240) if o: os.write(sys.stdout.fileno(), o) # restore tty settings back termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
My problem is a bit different as I wanted to collect both stdout and stderr from a running process, but ultimately the same since I wanted to render the output in a widget as its generated.
I did not want to resort to many of the proposed workarounds using Queues or additional Threads as they should not be necessary to perform such a common task as running another script and collecting its output.
After reading the proposed solutions and python docs I resolved my issue with the implementation below. Yes it only works for POSIX as I'm using the select function call.
I agree that the docs are confusing and the implementation is awkward for such a common scripting task. I believe that older versions of python have different defaults for Popen and different explanations so that created a lot of confusion. This seems to work well for both Python 2.7.12 and 3.5.2.
The key was to set bufsize=1 for line buffering and then universal_newlines=True to process as a text file instead of a binary which seems to become the default when setting bufsize=1 .
class workerThread(QThread): def __init__(self, cmd): QThread.__init__(self) self.cmd = cmd self.result = None ## return code self.error = None ## flag indicates an error self.errorstr = "" ## info message about the error def __del__(self): self.wait() DEBUG("Thread removed") def run(self): cmd_list = self.cmd.split(" ") try: cmd = subprocess.Popen(cmd_list, bufsize=1, stdin=None , universal_newlines=True , stderr=subprocess.PIPE , stdout=subprocess.PIPE) except OSError: self.error = 1 self.errorstr = "Failed to execute " + self.cmd ERROR(self.errorstr) finally: VERBOSE("task started...") import select while True: try: r,w,x = select.select([cmd.stdout, cmd.stderr],[],[]) if cmd.stderr in r: line = cmd.stderr.readline() if line != "": line = line.strip() self.emit(SIGNAL("update_error(QString)"), line) if cmd.stdout in r: line = cmd.stdout.readline() if line == "": break line = line.strip() self.emit(SIGNAL("update_output(QString)"), line) except IOError: pass cmd.wait() self.result = cmd.returncode if self.result < 0: self.error = 1 self.errorstr = "Task terminated by signal " + str(self.result) ERROR(self.errorstr) return if self.result: self.error = 1 self.errorstr = "exit code " + str(self.result) ERROR(self.errorstr) return return
ERROR, DEBUG and VERBOSE are simply macros that print output to the terminal.
This solution is IMHO 99.99% effective as it still uses the blocking readline function, so we assume the sub process is nice and outputs complete lines.
I welcome feedback to improve the solution as I am still new to Python.
This solution uses the select module to "read any available data" from an IO stream. This function blocks initially until data is available, but then reads only the data that is available and doesn't block further.
Given the fact that it uses the select module, this only works on Unix.
The code is fully PEP8-compliant.
import select def read_available(input_stream, max_bytes=None): """ Blocks until any data is available, then all available data is then read and returned. This function returns an empty string when end of stream is reached. Args: input_stream: The stream to read from. max_bytes (int|None): The maximum number of bytes to read. This function may return fewer bytes than this. Returns: str """ # Prepare local variables input_streams = [input_stream] empty_list = [] read_buffer = "" # Initially block for input using 'select' if len(select.select(input_streams, empty_list, empty_list)[0]) > 0: # Poll read-readiness using 'select' def select_func(): return len(select.select(input_streams, empty_list, empty_list, 0)[0]) > 0 # Create while function based on parameters if max_bytes is not None: def while_func(): return (len(read_buffer) < max_bytes) and select_func() else: while_func = select_func while True: # Read single byte at a time read_data = input_stream.read(1) if len(read_data) == 0: # End of stream break # Append byte to string buffer read_buffer += read_data # Check if more data is available if not while_func(): break # Return read buffer return read_buffer
Here is a module that supports non-blocking reads and background writes in python:
https://pypi.python.org/pypi/python-nonblock
Provides a function,
nonblock_read which will read data from the stream, if available, otherwise return an empty string (or None if the stream is closed on the other side and all possible data has been read)
You may also consider the python-subprocess2 module,
https://pypi.python.org/pypi/python-subprocess2
which adds to the subprocess module. So on the object returned from "subprocess.Popen" is added an additional method, runInBackground. This starts a thread and returns an object which will automatically be populated as stuff is written to stdout/stderr, without blocking your main thread.
请享用！
