Python如何连续填充子进程的多个线程？

Question

我正在 Linux 上运行一个应用程序 foo。在 Bash 脚本/终端提示符下，我的应用程序使用以下命令运行多线程：

$ foo -config x.ini -threads 4 < inputfile

系统监视器和顶部报告 foo 平均 CPU 负载约为 380%（四核计算机）。我在 Python 2.6x 中重新创建了此功能：

proc = subprocess.Popen("foo -config x.ini -threads 4", \
        shell=True, stdin=subprocess.PIPE, \
        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
mylist = ['this','is','my','test','app','.']
for line in mylist:
    txterr = ''
    proc.stdin.write(line.strip()+'\n')
    while not proc.poll() and not txterr.count('Finished'):
        txterr += subproc.stderr.readline()
    print proc.stdout.readline().strip(),

Foo 运行速度较慢，top 报告 CPU 负载为 100%。 Foo 在 shell=False 时也运行良好，但仍然很慢：

proc = subprocess.Popen("foo -config x.ini -threads 4".split(), \
        shell=False, stdin=subprocess.PIPE, \
        stdout=subprocess.PIPE, stderr=subprocess.PIPE)

有没有办法让 Python 子进程连续填充所有线程？

Answer 1

当您像这样使用 Popen 调用命令时，它是从 Python 还是从 shell 调用并不重要。启动进程的是“foo”命令，而不是 Python。

所以答案是“是的，从 Python 调用时子进程可以是多线程的。”

Answer 2

首先，您是否猜测它是单线程只是因为它使用了 100% 而不是 400% 的 CPU？

最好使用 top 程序检查已启动的线程数量，按 H 键显示线程。或者，使用 ps -eLf 并确保 NLWP 列显示多个线程。

Linux 与 CPU 的亲和力可能会非常不稳定；默认情况下，调度程序不会将进程移离它最后使用的处理器。这意味着，如果程序的所有四个线程都在单个处理器上启动，它们将永远共享该处理器。您必须使用诸如 taskset(1) 之类的工具来强制必须在单独的处理器上长时间运行的进程上的 CPU 关联性。例如，taskset -p-c 0 ;任务集 -p-c 1 ;任务集 -p-c 2 ；任务集 -p-c 3。

您可以使用 taskset -p 检索关联性，以了解当前设置的关联性。

（有一天，我想知道为什么我的 Folding At Home 进程使用的 CPU 时间比我预期的要少得多，我发现该死的调度程序将三个 FaH 任务放在一个超线程兄弟上，并将第四个 FaH 任务放在另一个 HT 兄弟上相同的核心。其他三个处理器闲置（第一个核心也运行得很热，其他三个核心冷四五度。呵呵。））

Answer 3

如果你的 python 脚本没有足够快地为 foo 进程提供数据，那么你可以将读取 stdout、stderr 的任务卸载到线程：

from Queue import Empty, Queue
from subprocess import PIPE, Popen
from threading import Thread

def start_thread(target, *args):
    t = Thread(target=target, args=args)
    t.daemon = True
    t.start()
    return t

def signal_completion(queue, stderr):
    for line in iter(stderr.readline, ''):
        if 'Finished' in line:
           queue.put(1) # signal completion
    stderr.close()

def print_stdout(q, stdout):
    """Print stdout upon receiving a signal."""
    text = []
    for line in iter(stdout.readline, ''):
        if not q.empty():
           try: q.get_nowait()               
           except Empty:
               text.append(line) # queue is empty
           else: # received completion signal              
               print ''.join(text),
               text = []
               q.task_done()
        else: # buffer stdout until the task is finished
            text.append(line)
    stdout.close()
    if text: print ''.join(text), # print the rest unconditionally

queue = Queue()
proc = Popen("foo -config x.ini -threads 4".split(), bufsize=1,
             stdin=PIPE, stdout=PIPE, stderr=PIPE)
threads =  [start_thread(print_stdout, queue, proc.stdout)]
threads += [start_thread(signal_completion, queue, proc.stderr)]

mylist = ['this','is','my','test','app','.']
for line in mylist:
    proc.stdin.write(line.strip()+'\n')
proc.stdin.close()
proc.wait()
for t in threads: t.join() # wait for stdout