Python: thực hiện mèo subprocess song song

Question

Tôi đang chạy nhiều cat | zgrep lệnh trên một máy chủ từ xa và thu thập sản lượng của họ riêng để chế biến tiếp:

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) 

này, tuy nhiên, kết quả trong thực hiện tuần tự của các tiến trình con ('ssh ... lệnh ...). Đỉnh thứ hai chờ đầu tiên kết thúc và vân vân.

Làm cách nào để tôi có thể sửa đổi mã này để các cuộc gọi subprocess chạy song song, trong khi vẫn có thể thu thập đầu ra cho từng cá nhân?

Answer 1

Một cách tiếp cận (chứ không phải là gợi ý khác đưa các quy trình shell trong nền) là sử dụng multithreading.

Các run phương pháp mà bạn có sau đó sẽ làm một cái gì đó như thế này:

thread.start_new_thread (myFuncThatDoesZGrep) 

Thu kết quả, bạn có thể làm một cái gì đó như thế này:

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 = [] 

Chạy chuỗi như đã nêu ở trên trong liên kết trên multithr eading. Khi đối tượng thread của bạn có myThread.finished == True, thì bạn có thể thu thập kết quả qua myThread.results.

Answer 2

Bạn không cần không multiprocessing cũng không threading để chạy subprocesses song song ví dụ .:

#!/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] 

nó chạy lệnh 5 vỏ cùng một lúc. Lưu ý: không phải chủ đề nào cũng không phải mô-đun multiprocessing được sử dụng tại đây. Không có điểm để thêm dấu và & vào các lệnh shell: Popen không chờ lệnh hoàn tất. Bạn cần gọi số .wait() một cách rõ ràng.

Đó là thuận lợi nhưng nó không phải là cần thiết để sử dụng chủ đề để thu thập kết quả từ subprocesses:

#!/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) 

liên quan: Python threading multiple bash subprocesses?.

Dưới đây là mã ví dụ mà được sản lượng từ vài subprocesses đồng thời trong cùng một 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()