Sottoprocesso Python in parallelo

Question

Desidero eseguire più processi in parallelo con la possibilità di eseguire stdout in qualsiasi momento. Come dovrei farlo? Devo eseguire thread per ogni chiamata subprocess.Popen(), cosa?

Answer 1

È possibile farlo in una singola discussione.

Supponiamo di avere uno script che stampa le linee a volte casuali:

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

e desideri per raccogliere l'uscita non appena sarà disponibile, è possibile utilizzare select su sistemi POSIX come @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 soluzione più portatile (che dovrebbe funzionare su Windows, Linux, OSX) possono utilizzare le discussioni lettore per ogni processo, vedere Non-blocking read on a subprocess.PIPE in python.

Ecco os.pipe() soluzione basata su che funziona su Unix e 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

Non è necessario eseguire un thread per ogni processo. È possibile dare un'occhiata agli stream stdout per ogni processo senza bloccarli e leggerli solo se sono disponibili dati da leggere.

Si do fare attenzione a non bloccarsi accidentalmente su di essi, però, se non si intende.

Answer 3

È possibile anche raccogliere stdout da più sottoprocessi in concomitanza con 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() 

Vedi Using Processes in Twisted.