Python多处理 - 发送输入到子进程

Question

我在python中使用多处理模块来并行启动几个进程。这些过程彼此独立。他们生成自己的输出并将结果写入不同的文件。每个进程使用subprocess.call方法调用外部工具。 它工作正常，直到我发现外部工具中的问题，由于某些错误情况，它进入'提示'模式并等待用户输入。现在在我的Python脚本中，我使用join方法等待所有进程完成其任务。这导致整个事情等待这个错误的子进程调用。我可以为每个进程设置一个超时时间，但是我不知道每个进程会持续多久，因此这个选项被排除了。

如何判断是否有子进程正在等待用户输入，以及如何向其发送“退出”命令？任何指向Python相关模块的指针或建议都会非常感谢。

我的代码在这里：

import subprocess 
import sys 
import os 
import multiprocessing 

def write_script(fname,e): 
    f = open(fname,'w') 
    f.write("Some useful cammnd calling external tool") 
    f.close() 
    subprocess.call(['chmod','+x',os.path.abspath(fname)]) 
    return os.path.abspath(fname) 

def run_use(mname,script): 
    print "ssh "+mname+" "+script 
    subprocess.call(['ssh',mname,script]) 

if __name__ == '__main__': 
    dict1 = {} 
    dict['mod1'] = ['pp1','ext2','les3','pw4'] 
    dict['mod2'] = ['aaa','bbb','ccc','ddd'] 
    machines = ['machine1','machine2','machine3','machine4'] 
    log_file.write(str(dict1.keys())) 
    for key in dict1.keys(): 
     arr = [] 
     for mod in dict1[key]: 
      d = {} 
      arr.append(mod) 
      if ((mod == dict1[key][-1]) | (len(arr)%4 == 0)): 
       for i in range(0,len(arr)): 
        e = arr.pop() 
        script = write_script(e+"_temp.sh",e) 
        d[i] = multiprocessing.Process(target=run_use,args=(machines[i],script,)) 
        d[i].daemon = True 
       for pp in d: 
        d[pp].start() 
       for pp in d: 
        d[pp].join() 

Answer 1

既然你写一个shell脚本来运行你的子，可你只要告诉他们从/dev/null读取输入？

#!/bin/bash 
# ... 
my_other_command -a -b arg1 arg2 < /dev/null 
# ... 

这可能会阻止他们阻止输入，是一个非常简单的解决方案。如果这不适合你，请阅读其他一些选项。

subprocess.call()函数只是简单地构造一个subprocess.Popen实例，然后调用wait()方法。因此，您的备用进程可以创建自己的subprocess.Popen实例，并在对象上使用poll()方法轮询它们，而不是使用wait()（在具有适当延迟的循环中）。这使得他们可以自由地与主进程保持通信，例如，您可以让主进程告诉子进程用terminate()或方法终止Popen实例，然后退出。

所以，问题是子进程如何判断子进程是否在等待用户输入，这是一个棘手的问题。我会说最简单的方法是监视子进程的输出并搜索用户输入提示符，假设它总是使用一些可以查找的字符串。或者，如果子进程预计会持续生成输出，那么您可以简单地查找任何输出，并且如果配置的时间量过去而没有任何输出，那么您声明该进程死了并终止它，如上所述。

由于您正在读取输出，实际上您不需要poll()或wait() - 关闭其输出文件描述符的过程足以知道在这种情况下它已终止。

这里的一个改性run_use()方法，其手表子进程的输出的一个示例：

def run_use(mname,script): 
    print "ssh "+mname+" "+script 
    proc = subprocess.Popen(['ssh',mname,script], stdout=subprocess.PIPE) 
    for line in proc.stdout: 
     if "UserPrompt>>>" in line: 
      proc.terminate() 
      break 

在这个例子中，我们假定该方法是要么被挂在上UserPrompt>>>（与适当的字符串替换） 或它自然终止。例如，如果它陷入了无限循环，那么你的脚本仍然不会终止 - 你只能通过整体超时来解决这个问题，但你似乎并不热衷于这样做。不过，希望你的子过程不会以这种方式行事。

最后，如果您事先不知道将从您的流程中提供的提示，那么您的工作相当困难。实际上，你所要做的就是监视一个外部进程，并知道它何时被阻塞在文件描述符上读取，我不相信有一个特别干净的解决方案。你可能考虑运行strace或类似的过程，但这是一个相当可怕的黑客攻击，我真的不会推荐它。诸如strace之类的东西对手动诊断非常有用，但它们确实不应该是生产设置的一部分。