Python: eseguire gatto sottoprocesso in parallelo

Question

Sono in esecuzione diverse cat | zgrep comandi su un server remoto e raccogliendo la loro produzione individualmente per ulteriori elaborazioni:

class MainProcessor(mp.Process): 
    def __init__(self, peaks_array): 
     super(MainProcessor, self).__init__() 
     self.peaks_array = peaks_array 

    def run(self): 
     for peak_arr in self.peaks_array: 
      peak_processor = PeakProcessor(peak_arr) 
      peak_processor.start() 

class PeakProcessor(mp.Process): 
    def __init__(self, peak_arr): 
     super(PeakProcessor, self).__init__() 
     self.peak_arr = peak_arr 

    def run(self): 
     command = 'ssh remote_host cat files_to_process | zgrep --mmap "regex" ' 
     log_lines = (subprocess.check_output(command, shell=True)).split('\n') 
     process_data(log_lines) 

Ciò, tuttavia, comporta l'esecuzione sequenziale del sottoprocesso ('SSH ... gatto ... ') comandi. Il secondo picco attende il primo alla fine e così via.

Come posso modificare questo codice in modo che le chiamate di sottoprocesso vengano eseguite in parallelo, pur essendo in grado di raccogliere l'output per ciascuna individualmente?

Answer 1

Un altro approccio (piuttosto che l'altro suggerimento di mettere i processi di shell in background) è quello di utilizzare multithreading.

Il run metodo che avete avrebbe poi fare qualcosa del genere:

thread.start_new_thread (myFuncThatDoesZGrep) 

Per raccogliere risultati, è possibile fare qualcosa di simile:

class MyThread(threading.Thread): 
    def run(self): 
     self.finished = False 
     # Your code to run the command here. 
     blahBlah() 
     # When finished.... 
     self.finished = True 
     self.results = [] 

Eseguire la discussione come indicato sopra nel collegamento su multithr eading. Quando il tuo oggetto thread ha myThread.finished == True, puoi raccogliere i risultati tramite myThread.results.

Answer 2

Non è necessario né multiprocessing né threading per eseguire sottoprocessi in parallelo ad es .:

#!/usr/bin/env python 
from subprocess import Popen 

# run commands in parallel 
processes = [Popen("echo {i:d}; sleep 2; echo {i:d}".format(i=i), shell=True) 
      for i in range(5)] 
# collect statuses 
exitcodes = [p.wait() for p in processes] 

corre comanda 5 shell contemporaneamente. Nota: qui non vengono utilizzati né i thread né il modulo multiprocessing. Non è necessario aggiungere la e commerciale & ai comandi della shell: Popen non attende il completamento del comando. Devi chiamare lo .wait() in modo esplicito.

È conveniente, ma non è necessario usare fili per raccogliere uscita da sottoprocessi:

#!/usr/bin/env python 
from multiprocessing.dummy import Pool # thread pool 
from subprocess import Popen, PIPE, STDOUT 

# run commands in parallel 
processes = [Popen("echo {i:d}; sleep 2; echo {i:d}".format(i=i), shell=True, 
        stdin=PIPE, stdout=PIPE, stderr=STDOUT, close_fds=True) 
      for i in range(5)] 

# collect output in parallel 
def get_lines(process): 
    return process.communicate()[0].splitlines() 

outputs = Pool(len(processes)).map(get_lines, processes) 

correlati: Python threading multiple bash subprocesses?.

Ecco esempio di codice che viene emesso dal più operazioni contemporaneamente nello stesso thread:

#!/usr/bin/env python3 
import asyncio 
import sys 
from asyncio.subprocess import PIPE, STDOUT 

@asyncio.coroutine 
def get_lines(shell_command): 
    p = yield from asyncio.create_subprocess_shell(shell_command, 
      stdin=PIPE, stdout=PIPE, stderr=STDOUT) 
    return (yield from p.communicate())[0].splitlines() 

if sys.platform.startswith('win'): 
    loop = asyncio.ProactorEventLoop() # for subprocess' pipes on Windows 
    asyncio.set_event_loop(loop) 
else: 
    loop = asyncio.get_event_loop() 

# get commands output in parallel 
coros = [get_lines('"{e}" -c "print({i:d}); import time; time.sleep({i:d})"' 
        .format(i=i, e=sys.executable)) for i in range(5)] 
print(loop.run_until_complete(asyncio.gather(*coros))) 
loop.close()