Python多处理:只有一个进程正在运行

Question

我正在尝试使用Python多处理模块生成多个并行进程.基本上,我做了类似的事情

pool = Pool(30)
results = [pool.apply_async(foo, (trainData, featureVector, terms, selLabel)) for selLabel in selLabels]
for r in results:
    tmp = r.get()
    modelFiles[tmp[0]] = tmp[1]

产生了30个进程,但是,似乎大多数进程已进入休眠状态,而实际只有一个进程正在运行.以下是我从ps得到的:

PID %CPU %MEM    VSZ   RSS TTY      STAT START   TIME COMMAND

31267 74.6  2.4 7125412 6360080 pts/1 Sl+  13:06  24:25  \_ python2.6 /home/PerlModules/Python/DoOVA.py

31427 27.4  2.3 6528532 6120904 pts/1 R+   13:20   5:18      \_ python2.6 /home/PerlModules/Python/DoOVA.py

31428  0.0  1.3 4024724 3617016 pts/1 S+   13:20   0:00      \_ python2.6 /home/PerlModules/Python/DoOVA.py

31429  0.0  1.3 4024724 3617016 pts/1 S+   13:20   0:00      \_ python2.6 /home/PerlModules/Python/DoOVA.py

31430  0.0  1.3 4024724 3617016 pts/1 S+   13:20   0:00      \_ python2.6 /home/PerlModules/Python/DoOVA.py

DoOVA.py是我正在运行的脚本.他们中的大多数都有一个地位S+.

谁能给我一些关于问题的线索？我知道输入争论的featureVector规模相当大,比如大约300MB.那会是个问题吗？我运行的机器有几TB的内存.

foo做的事情如下:

def foo(trainData, featureVector, terms, selLabel, penalty):
    outputFile = 'train_'+selLabel+'.dat'
    annotation = dict()
    for id in trainData:
        if trainData[id] == selLabel:
            annotation[id] = '1'
        else:
            annotation[id] = '-1'
    try:
        os.mkdir(selLabel)
        os.chdir(selLabel)
    except OSError:
        os.chdir(selLabel)
    ###Some more functions, which involves a command line call through call from subprocess module
    os.chdir('../')
    return (selLabel, 'SVM_' + selLabel + '.model')

所有其他输入参数的大小都很小.机器至少有100 cpus.在每次运行中,即使在创建任何目录之前,脚本也需要很长时间,尽管在os.mkdir()之前foo中没有发生重大计算.

Answer 1

由于注释表明你想featureVector使用initializer和initargs参数传递Pool.在Unix类型的系统上,这将导致大量的性能提升(即使只有1项selLabels),因为该值将基本上免费使用传递给子进程os.fork.否则,每次foo调用时,featureVector将由父进程进行pickle,通过管道传递并由子进程进行unpickled.这将花费很长时间,并且基本上将序列化所有子进程,因为它们将等待父进程挑选并featureVector逐个发送每个调用的副本.

由于对于我上面谈论的内容存在一些困惑,所以这里有一个更长的解释,说明代码中发生的内容与当前编写的内容有关:

创建Pool对象时,将立即创建30个工作进程,即创建该Pool对象的主进程的所有子进程.为了与每个子进程进行通信,创建了一个管道.此管道允许父进程和子进程之间的双向通信.父级使用管道来指示子进程执行的操作,子级使用管道通知父级任何操作的结果.

当您第一次调用pool.apply_async父进程时,通过管道发送命令,指示子进程foo使用提供的参数执行该函数.由于其中一个论点是巨大的,300MB,这最终需要很长时间.父进程必须pickle对象.这会将对象(及其引用的所有内容)转换为可通过管道发送的字节流.

由于管道只能容纳大约64k(Linux默认值),并且您发送的内容远不止这些,因此可以有效地同步父进程和其中一个子进程.父进程只能以子进程可以接收和取消它们的速度发送参数,并且子进程只能像父进程一样快地接收参数并发送它们.虽然这是在进行所有其他子进程必须等待.父进程一次只能向一个子进程发送命令.

一旦父进程完成了第一次调用的所有参数的发送foo,它就可以继续发送命令再次调用foo.在此之后不久,一旦子进程收到所有参数,孩子就会打电话foo.(这就是为什么在创建任何目录之前需要很长时间,foo甚至在调用之前需要很长时间.)foo返回之后,子进程将等待父进程发送另一个命令.如果foo自己花费足够短的时间来执行,则接收第一个要调用的命令的相同子进程也可能foo会收到第二个要调用的命令foo.

除非foo它本身需要很长时间才能执行,只要比featureVector通过管道发送所花费的时间长或长,否则你将被有效地限制为只执行一个子进程.父进程将尝试命令子进程foo尽可能快地调用,但由于featureVector它太大,它只能以非常慢的速率执行.一旦完成将命令发送到一个进程进行调用foo,它命令调用的先前进程很久以前foo就已经完成了调用foo.运行子进程之间几乎没有重叠.

为了解决代码中的性能问题,您需要执行以下操作:

def child_initialize(_trainData, _featureVector, _terms):
     global trainData, featureVector, terms
     trainData = _trainData
     featureVector = _featureVector
     terms = _terms

def foo(selLabel):
     ...

pool = Pool(30, initialize = child_initialize, initargs = (trainData, featureVector, terms))
results = [pool.apply_async(foo, (selLabel,)) for selLabel in selLabels]

此代码也通过trainData并term使用initargs它们也不会改变.

虽然这会导致巨大的性能提升,并允许子进程并行运行,但这并不一定意味着子进程ps将更频繁地出现在可运行状态.您的示例foo函数看起来似乎将花费大部分时间等待"命令行调用"完成.