Subproceso de Python en paralelo

Question

Quiero ejecutar muchos procesos en paralelo con la capacidad de tomar stdout en cualquier momento. ¿Cómo debería hacerlo? ¿Debo ejecutar hilo para cada llamada subprocess.Popen(), un qué?

Answer 1

Puede hacerlo en un solo hilo.

Suponga que tiene una secuencia de comandos que imprime líneas en momentos aleatorios:

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

y que le gustaría recoger la salida tan pronto como esté disponible, se puede utilizar en sistemas POSIX select como @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 

Una solución más portátil (que debería funcionar en Windows, Linux, OS X) pueden utilizar hilos lector para cada proceso, consulte Non-blocking read on a subprocess.PIPE in python.

Aquí es os.pipe() solución basada en que funciona en Unix y 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

No necesita ejecutar un hilo para cada proceso. Puede echar un vistazo a las transmisiones stdout para cada proceso sin bloquearlas, y solo leer de ellas si tienen datos disponibles para leer.

Usted haga tenga cuidado de no bloquearlos accidentalmente, sin embargo, si no tiene intención de hacerlo.

Answer 3

También puede obtener la salida estándar a partir de múltiples subprocesos simultáneamente usando 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() 

Ver Using Processes in Twisted.