Mimicing glib.spawn_async z Popen ...

Question

Funkcja glib.spawn_async umożliwia podpięcie trzech wywołań zwrotnych wywoływanych przy zdarzeniu na stdout, stderr oraz po zakończeniu procesu.

Jak mogę naśladować tę samą funkcjonalność z subprocess z wątkami lub asynio?

Bardziej interesuje mnie funkcjonalność niż gwintowanie/asynio, ale odpowiedź, która zawiera obie, przyniesie nagrodę.

Oto program zabawka, która pokazuje, co chcę zrobić:

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('') 

Dodam, że od readlines() blokuje, powyższe będzie buforować wszystkie wyjścia i wysłać go na raz. Jeśli nie tego chcesz, musisz użyć readline() i upewnić się, że po zakończeniu polecenia skończyłeś czytać wszystkie linie, których wcześniej nie czytałeś.

Answer 1

asyncio ma subprocess_exec, nie ma potrzeby, aby korzystać z modułu subprocess w ogóle:

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() 

Z podproces i gwintowania, to proste, jak również. Można po prostu tarła wątku za rury i jeden do wait() dla procesu:

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() 

Jednak umysł GLib robi nie używać wątków do asynchroneously wykonać swoje zwrotnych. Używa pętli zdarzeń, tak jak robi to asyncio. Chodzi o to, że rdzeniem twojego programu jest pętla, która czeka, aż coś się stanie, a następnie synchronicznie wykona skojarzone wywołanie zwrotne. W twoim przypadku jest to "dane stają się dostępne do odczytu na jednej z rur" i "podproces został zakończony". Ogólnie rzecz biorąc, jest to również coś takiego jak "ruchy myszy zgłaszane przez serwer X11", "nadchodzi ruch sieciowy" itp. Możesz emulować zachowanie glib, pisząc własną pętlę zdarzeń. Użyj select module na dwóch rurach. Jeśli opcja select zgłosi, że rury są czytelne, ale read nie zwraca żadnych danych, proces prawdopodobnie zakończył się - w tym przypadku wywołaj metodę poll() w obiekcie podprocesu, aby sprawdzić, czy jest ona zakończona i wywołaj wywołanie zwrotne wyjścia, jeśli tak się stało, lub błąd callback else.