Ottieni output da un processo di sottoprocesso python a tornado

Question

Ho cercato molto e non ho trovato come ottengo l'output da un sottoprocesso Python in esecuzione in Tornado. Quello che voglio è qualcosa di simile allo Travis CI. In una pagina di amministrazione inizierò il lavoro, il server riceverà la richiesta e avvierà un sottoprocesso. Questo sottoprocesso eseguirà un data mining e alimenterà un buffer di stringa con alcuni registri. Otterrò questo registro con un po 'di ajax con settimeout o un websocket e trasmetterò questi log nella pagina. Anche se l'utente chiude la pagina e ci torna più tardi, ci saranno i log che normalmente aggiorneranno. Bene, è davvero molto simile a Travis.

Answer 1

Questo post del blog mostra un modo di fare questo: http://stefaanlippens.net/python-asynchronous-subprocess-pipe-reading

Essenzialmente il post mostra come prevenire stallo durante la lettura l'output di un processo in modo asincrono la lettura sia stdout e stderr. È possibile sostituire il comando producer da __main__ per eseguire qualsiasi comando desiderato e le istruzioni di stampa con codice per gestire l'output in Tornado.

Aggiornamento: ho incluso quanto segue nel caso in cui il blog viene rimosso a:

... cosa succede se si desidera leggere lo standard output e la linea di errore per riga, ad esempio perché si desidera monitorare un processo più lungo? Su il web si possono trovare molte soluzioni, con vari gradi di complessità, astrazione e dipendenze . Una soluzione (con codice limitato e senza dipendenze al di fuori della libreria standard) consiste nel leggere le pipe in thread separati, quindi una pipe non può bloccarne un'altra.

Il codice seguente mostra un'implementazione di esempio. Lo script è impostato su in modo tale da essere utilizzato sia per il genitore che come processo secondario.

Per il processo figlio: quando chiamato con argomento 'produce', esegue la funzione produce() che visualizza solo alcune righe casualmente sullo standard e un errore standard. Tra le linee c'è un tocco di ritardo simulare un processo più lungo in esecuzione. Il processo genitore (script chiamato senza argomenti), implementato nella funzione di consumo(), richiama lo stesso script in "modalità figlio" come sottoprocesso e monitora l'output riga per riga, senza sapere in anticipo da quale pipe ogni linea verrà venire.

La classe AsynchronousFileReader è per i thread che leggeranno lo standard output e le pipe di errore in modo asincrono e inseriscono ciascuna riga in una coda . Il thread principale può quindi monitorare il sottoprocesso guardando le righe mentre arrivano sulle code.

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