Pobierz dane wyjściowe z zadania podprogramu Pythona do tornada

Question

Szukałem dużo i nie znalazłem sposobu, w jaki otrzymuję dane wyjściowe z działającego podprocesora Pythona do Tornado. To, czego chcę, to coś w stylu Travis CI. Na stronie administratora rozpocznę pracę, serwer odbierze żądanie i uruchomi podproces. Podproces ten wykona eksplorację danych i doda bufor ciągów do niektórych dzienników. Dostaję te logi z jakimś ajaxem z settimeout lub websocket i wypiszę to logując się na stronie. Nawet jeśli użytkownik zamknie stronę i powróci do niej później, będą to dzienniki i zazwyczaj będą się aktualizować. Cóż, jest bardzo podobny do Travisa.

Answer 1

Ten blogu pokazuje sposób to zrobić: http://stefaanlippens.net/python-asynchronous-subprocess-pipe-reading

Zasadniczo post pokazuje, jak uniknąć impasu podczas czytania wyjście procesu przez asynchroniczny odczyt zarówno stdout i stderr. Możesz zastąpić komendę producer z __main__, aby uruchomić dowolne polecenie i instrukcje drukowania z kodem do obsługi wyjścia w Tornado.

Aktualizacja: mam włączone następuje w przypadku, gdy blog zostanie zdjęty:

... co, jeśli chcesz przeczytać standardowe wyjście błędu i linię po linii, na przykład z powodu chcesz monitorować dłuższy proces? Na stronie internetowej można znaleźć wiele rozwiązań, o różnym stopniu złożoności, abstrakcji i zależności. Jednym rozwiązaniem (z ograniczonym kodem i bez zależności poza biblioteką standardową) jest odczytanie rur w oddzielnych wątkach, więc jedna rura nie może zablokować drugiej.

Poniższy kod przedstawia przykładową implementację. Skrypt został skonfigurowany w taki sposób, aby był używany zarówno dla rodzica jako procesu potomnego.

Dla procesu potomnego: po wywołaniu z argumentem "produkcji", uruchamia funkcję produkującą(), która po prostu renderuje niektóre linie losowo na standardowym wyjściu i standardowym błędzie . Pomiędzy liniami pojawia się dotyk opóźnienia, który symuluje dłuższy proces. Proces macierzysty (skrypt wywoływany bez argumentów), zaimplementowany w funkcji consume(), wywołuje ten sam skrypt w "trybie potomnym" jako podprogram i monitoruje linię wyjściową po linii, nie wiedząc w numerze z góry, z której rury każda linia będzie chodź.

Klasa AsynchronousFileReader jest przeznaczona dla wątków, które asynchronicznie odczytują standardowe wyjścia i przewody błędów i umieszczają każdą linię w kolejce . Główny wątek może następnie monitorować podproces, obserwując linie , gdy pojawiają się w kolejkach.

import sys 
import subprocess 
import random 
import time 
import threading 
import Queue 

class AsynchronousFileReader(threading.Thread): 
    ''' 
    Helper class to implement asynchronous reading of a file 
    in a separate thread. Pushes read lines on a queue to 
    be consumed in another thread. 
    ''' 

    def __init__(self, fd, queue): 
     assert isinstance(queue, Queue.Queue) 
     assert callable(fd.readline) 
     threading.Thread.__init__(self) 
     self._fd = fd 
     self._queue = queue 

    def run(self): 
     '''The body of the tread: read lines and put them on the queue.''' 
     for line in iter(self._fd.readline, ''): 
      self._queue.put(line) 

    def eof(self): 
     '''Check whether there is no more content to expect.''' 
     return not self.is_alive() and self._queue.empty() 

def consume(command): 
    ''' 
    Example of how to consume standard output and standard error of 
    a subprocess asynchronously without risk on deadlocking. 
    ''' 

    # Launch the command as subprocess. 
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) 

    # Launch the asynchronous readers of the process' stdout and stderr. 
    stdout_queue = Queue.Queue() 
    stdout_reader = AsynchronousFileReader(process.stdout, stdout_queue) 
    stdout_reader.start() 
    stderr_queue = Queue.Queue() 
    stderr_reader = AsynchronousFileReader(process.stderr, stderr_queue) 
    stderr_reader.start() 

    # Check the queues if we received some output (until there is nothing more to get). 
    while not stdout_reader.eof() or not stderr_reader.eof(): 
     # Show what we received from standard output. 
     while not stdout_queue.empty(): 
      line = stdout_queue.get() 
      print 'Received line on standard output: ' + repr(line) 

     # Show what we received from standard error. 
     while not stderr_queue.empty(): 
      line = stderr_queue.get() 
      print 'Received line on standard error: ' + repr(line) 

     # Sleep a bit before asking the readers again. 
     time.sleep(.1) 

    # Let's be tidy and join the threads we've started. 
    stdout_reader.join() 
    stderr_reader.join() 

    # Close subprocess' file descriptors. 
    process.stdout.close() 
    process.stderr.close() 

def produce(items=10): 
    ''' 
    Dummy function to randomly render a couple of lines 
    on standard output and standard error. 
    ''' 
    for i in range(items): 
     output = random.choice([sys.stdout, sys.stderr]) 
     output.write('Line %d on %s\n' % (i, output)) 
     output.flush() 
     time.sleep(random.uniform(.1, 1)) 

if __name__ == '__main__': 
    # The main flow: 
    # if there is an command line argument 'produce', act as a producer 
    # otherwise be a consumer (which launches a producer as subprocess). 
    if len(sys.argv) == 2 and sys.argv[1] == 'produce': 
     produce(10) 
    else: 
     consume(['python', sys.argv[0], 'produce'])