Podproces Pythona równolegle

Question

Chcę uruchamiać wiele procesów równolegle ze zdolnością do przyjmowania stdout w dowolnym momencie. Jak mam to zrobić? Czy muszę uruchomić wątek dla każdego wywołania subprocess.Popen(), co?

Answer 1

Możesz to zrobić w jednym wątku.

Załóżmy, że masz skrypt, który drukuje wiersze losowo razy:

#!/usr/bin/env python 
#file: child.py 
import os 
import random 
import sys 
import time 

for i in range(10): 
    print("%2d %s %s" % (int(sys.argv[1]), os.getpid(), i)) 
    sys.stdout.flush() 
    time.sleep(random.random()) 

i chcesz zbierać wyjścia, jak tylko stanie się ona dostępna, można użyć select na systemach POSIX jak @zigg suggested:

#!/usr/bin/env python 
from __future__ import print_function 
from select  import select 
from subprocess import Popen, PIPE 

# start several subprocesses 
processes = [Popen(['./child.py', str(i)], stdout=PIPE, 
        bufsize=1, close_fds=True, 
        universal_newlines=True) 
      for i in range(5)] 

# read output 
timeout = 0.1 # seconds 
while processes: 
    # remove finished processes from the list (O(N**2)) 
    for p in processes[:]: 
     if p.poll() is not None: # process ended 
      print(p.stdout.read(), end='') # read the rest 
      p.stdout.close() 
      processes.remove(p) 

    # wait until there is something to read 
    rlist = select([p.stdout for p in processes], [],[], timeout)[0] 

    # read a line from each process that has output ready 
    for f in rlist: 
     print(f.readline(), end='') #NOTE: it can block 

bardziej przenośnym rozwiązaniem (które powinny działać w systemie Windows, Linux, OSX) mogą wykorzystywać wątki reader dla każdego procesu, zobacz Non-blocking read on a subprocess.PIPE in python.

Oto os.pipe() -na rozwiązanie, które działa na systemach Unix i Windows:

#!/usr/bin/env python 
from __future__ import print_function 
import io 
import os 
import sys 
from subprocess import Popen 

ON_POSIX = 'posix' in sys.builtin_module_names 

# create a pipe to get data 
input_fd, output_fd = os.pipe() 

# start several subprocesses 
processes = [Popen([sys.executable, 'child.py', str(i)], stdout=output_fd, 
        close_fds=ON_POSIX) # close input_fd in children 
      for i in range(5)] 
os.close(output_fd) # close unused end of the pipe 

# read output line by line as soon as it is available 
with io.open(input_fd, 'r', buffering=1) as file: 
    for line in file: 
     print(line, end='') 
# 
for p in processes: 
    p.wait() 

Answer 2

Nie trzeba uruchamiać wątku dla każdego procesu. Możesz przeglądać strumienie stdout dla każdego procesu bez blokowania ich i czytać tylko z nich, jeśli mają dane do odczytania.

Musisz być ostrożny, aby nie zablokować ich przypadkowo, jeśli nie masz na to ochoty.

Answer 3

Można również zbierać stdout z wielu podprocesów jednocześnie stosując twisted:

#!/usr/bin/env python 
import sys 
from twisted.internet import protocol, reactor 

class ProcessProtocol(protocol.ProcessProtocol): 
    def outReceived(self, data): 
     print data, # received chunk of stdout from child 

    def processEnded(self, status): 
     global nprocesses 
     nprocesses -= 1 
     if nprocesses == 0: # all processes ended 
      reactor.stop() 

# start subprocesses 
nprocesses = 5 
for _ in xrange(nprocesses): 
    reactor.spawnProcess(ProcessProtocol(), sys.executable, 
         args=[sys.executable, 'child.py'], 
         usePTY=True) # can change how child buffers stdout 
reactor.run() 

Zobacz Using Processes in Twisted.