Podproces Pythona równolegle

Question

Podproces Pythona równolegle

Chcę uruchomić wiele procesów równolegle z możliwością podjęcia stdout w dowolnym momencie. Jak mam to zrobić? Czy muszę uruchamiać wątek dla każdego subprocess.Popen() wywołania, a co?

13

python subprocess

Author: sashab, 2012-03-17

Source

3 answers

Można również zbierać stdout z wielu podprocesów jednocześnie za pomocą 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 wykorzystanie procesów w Twisted .

6

Author: jfs,
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/doraprojects.net/template/agent.layouts/content.php on line 54
2012-03-17 01:11:24

Nie musisz uruchamiać wątku dla każdego procesu. Możesz zerknąć na strumienie stdout dla każdego procesu bez blokowania ich i odczytywać je tylko wtedy, gdy mają dostępne dane.

Musisz uważać, aby nie zablokować ich przypadkowo, jeśli nie masz zamiaru.

4

Author: Amber,
Warning: date(): Invalid date.timezone value 'Europe/Kyiv', we selected the timezone 'UTC' for now. in /var/www/agent_stack/data/www/doraprojects.net/template/agent.layouts/content.php on line 54
2012-03-16 20:12:27

score 14 · Accepted Answer

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

Załóżmy, że masz skrypt, który drukuje linie w losowych momentach:

#!/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 zebrać dane wyjściowe, gdy tylko będzie dostępne, możesz użyć select w systemach POSIX jako @Zigg zasugerował :

#!/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śne rozwiązanie (które powinno działać na Windows, Linux, OSX) może używać wątków czytnika dla każdego procesu, zobacz Non-blocking read on a subprocess.Rura w Pythonie .

Oto os.pipe()-rozwiązanie oparte na systemie 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()