Bắt chước glib.spawn_async với Popen…

Question

Chức năng glib.spawn_async cho phép bạn móc ba cuộc gọi lại được gọi trên sự kiện trên stdout, stderr và hoàn tất quá trình.

Làm cách nào tôi có thể bắt chước cùng một chức năng với subprocess bằng một trong hai chủ đề hoặc asyncio?

Tôi quan tâm nhiều hơn đến chức năng hơn là luồng/asynio nhưng câu trả lời chứa cả hai sẽ kiếm được tiền thưởng.

Dưới đây là một chương trình đồ chơi cho thấy những gì tôi muốn làm:

import glib 
import logging 
import os 
import gtk 


class MySpawn(object): 
    def __init__(self): 
     self._logger = logging.getLogger(self.__class__.__name__) 

    def execute(self, cmd, on_done, on_stdout, on_stderr): 
     self.pid, self.idin, self.idout, self.iderr = \ 
      glib.spawn_async(cmd, 
          flags=glib.SPAWN_DO_NOT_REAP_CHILD, 
          standard_output=True, 
          standard_error=True) 
     fout = os.fdopen(self.idout, "r") 
     ferr = os.fdopen(self.iderr, "r") 
     glib.child_watch_add(self.pid, on_done) 
     glib.io_add_watch(fout, glib.IO_IN, on_stdout) 
     glib.io_add_watch(ferr, glib.IO_IN, on_stderr) 
     return self.pid 


if __name__ == '__main__': 
    logging.basicConfig(format='%(thread)d %(levelname)s: %(message)s', 
         level=logging.DEBUG) 
    cmd = '/usr/bin/git ls-remote https://github.com/DiffSK/configobj'.split() 

    def on_done(pid, retval, *args): 
     logging.info("That's all folks!…") 

    def on_stdout(fobj, cond): 
     """This blocks which is fine for this toy example…""" 
     for line in fobj.readlines(): 
      logging.info(line.strip()) 
     return True 

    def on_stderr(fobj, cond): 
     """This blocks which is fine for this toy example…""" 
     for line in fobj.readlines(): 
      logging.error(line.strip()) 
     return True 

    runner = MySpawn() 
    runner.execute(cmd, on_done, on_stdout, on_stderr) 
    try: 
     gtk.main() 
    except KeyboardInterrupt: 
     print('') 

tôi nên thêm rằng kể từ khi readlines() được ngăn chặn, bên trên sẽ đệm tất cả các đầu ra và gửi nó cùng một lúc. Nếu đây không phải là điều bạn muốn, thì bạn phải sử dụng readline() và đảm bảo rằng khi kết thúc lệnh, bạn đọc xong tất cả các dòng bạn chưa đọc trước đó.

Answer 1

asyncio có subprocess_exec, không có nhu cầu sử dụng các mô-đun subprocess tại tất cả:

import asyncio 

class Handler(asyncio.SubprocessProtocol): 
    def pipe_data_received(self, fd, data): 
     # fd == 1 for stdout, and 2 for stderr 
     print("Data from /bin/ls on fd %d: %s" % (fd, data.decode())) 

    def pipe_connection_lost(self, fd, exc): 
     print("Connection lost to /bin/ls") 

    def process_exited(self): 
     print("/bin/ls is finished.") 

loop = asyncio.get_event_loop() 
coro = loop.subprocess_exec(Handler, "/bin/ls", "/") 

loop.run_until_complete(coro) 
loop.close() 

Với tiến trình con và luồng, nó đơn giản là tốt. Bạn chỉ có thể đẻ trứng một sợi mỗi ống, và một để wait() cho quá trình này:

import subprocess 
import threading 

class PopenWrapper(object): 
    def __init__(self, args): 
     self.process = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.DEVNULL) 

     self.stdout_reader_thread = threading.Thread(target=self._reader, args=(self.process.stdout,)) 
     self.stderr_reader_thread = threading.Thread(target=self._reader, args=(self.process.stderr,)) 
     self.exit_watcher = threading.Thread(target=self._exit_watcher) 

     self.stdout_reader_thread.start() 
     self.stderr_reader_thread.start() 
     self.exit_watcher.start() 

    def _reader(self, fileobj): 
     for line in fileobj: 
      self.on_data(fileobj, line) 

    def _exit_watcher(self): 
     self.process.wait() 
     self.stdout_reader_thread.join() 
     self.stderr_reader_thread.join() 
     self.on_exit() 

    def on_data(self, fd, data): 
     return NotImplementedError 

    def on_exit(self): 
     return NotImplementedError 

    def join(self): 
     self.process.wait() 

class LsWrapper(PopenWrapper): 
    def on_data(self, fd, data): 
     print("Received on fd %r: %s" % (fd, data)) 

    def on_exit(self): 
     print("Process exited.") 


LsWrapper(["/bin/ls", "/"]).join() 

Tuy nhiên, nhớ rằng GLib không không đề sử dụng để thực hiện asynchroneously callbacks của bạn. Nó sử dụng một vòng lặp sự kiện, giống như asyncio. Ý tưởng là ở cốt lõi của chương trình của bạn là một vòng lặp chờ đợi cho đến khi một cái gì đó xảy ra, và sau đó thực hiện đồng bộ một cuộc gọi lại liên quan. Trong trường hợp của bạn, đó là "dữ liệu có sẵn để đọc trên một trong các đường ống" và "quy trình con đã thoát". Nói chung, nó cũng giống như "máy chủ X11 báo cáo chuyển động của chuột", "có lưu lượng mạng đến", v.v. Bạn có thể mô phỏng hành vi của glib bằng cách viết vòng lặp sự kiện của riêng bạn. Sử dụng select module trên hai ống. Nếu chọn báo cáo rằng các đường ống có thể đọc được, nhưng read không trả về dữ liệu, quá trình có thể đã thoát - gọi phương thức poll() trên đối tượng subprocess trong trường hợp này để kiểm tra xem nó có hoàn thành hay không gọi ngược lại.