我正在使用多处理池来管理 tesseract 进程(缩微胶片的 OCRing 页)。通常在一个包含 20 个 tesseract 进程的池中,一些页面将更难进行 OCR,因此这些进程比其他进程花费的时间长得多。与此同时,池只是挂起,大部分 CPU 都没有被利用。我希望留下这些散乱者继续,但我也想启动更多进程来填充许多其他 CPU,这些 CPU 现在处于空闲状态,而这几个粘性页面正在完成。我的问题:有没有办法加载新进程来利用那些闲置的 CPU。换句话说,是否可以在等待整个池完成之前填充池中的空位?
我可以使用 starmap 的异步版本,然后在当前池减少到一定数量的事件进程时加载一个新池。但这似乎不雅。根据需要自动保持在流程中的插槽会更优雅。
这是我的代码现在的样子:
def getMpBatchMap(fileList, commandTemplate, concurrentProcesses):
mpBatchMap = []
for i in range(concurrentProcesses):
fileName = fileList.readline()
if fileName:
mpBatchMap.append((fileName, commandTemplate))
return mpBatchMap
def executeSystemProcesses(objFileName, commandTemplate):
objFileName = objFileName.strip()
logging.debug(objFileName)
objDirName = os.path.dirname(objFileName)
command = commandTemplate.substitute(objFileName=objFileName, objDirName=objDirName)
logging.debug(command)
subprocess.call(command, shell=True)
def process(FILE_LIST_FILENAME, commandTemplateString, concurrentProcesses=3):
"""Go through the list of files and run the provided command against them,
one at a time. Template string maps the terms $objFileName and $objDirName.
Example:
>>> runBatchProcess('convert -scale 256 "$objFileName" "$objDirName/TN.jpg"')
"""
commandTemplate = Template(commandTemplateString)
with open(FILE_LIST_FILENAME) as fileList:
while 1:
# Get a batch of x files to process
mpBatchMap = getMpBatchMap(fileList, commandTemplate, concurrentProcesses)
# Process them
logging.debug('Starting MP batch of %i' % len(mpBatchMap))
if mpBatchMap:
with Pool(concurrentProcesses) as p:
poolResult = p.starmap(executeSystemProcesses, mpBatchMap)
logging.debug('Pool result: %s' % str(poolResult))
else:
break
最佳答案
你在这里混淆了一些东西。该池始终保持一定数量的指定进程处于事件状态。只要您不关闭池,无论是手动关闭还是通过离开上下文管理器的 with-block,您都不需要用进程重新填充池,因为它们不会去任何地方。
您可能想说的是“任务”,这些进程可以处理的任务。任务是您传递给池方法的可迭代对象的每个进程 block 。是的,有一种方法可以使用池中的空闲进程处理新任务,在所有先前排队的任务都已处理完之前。您已经为此选择了正确的工具,即池方法的异步版本。您所要做的就是重新应用某种异步池方法。
from multiprocessing import Pool
import os
def busy_foo(x):
x = int(x)
for _ in range(x):
x - 1
print(os.getpid(), ' returning: ', x)
return x
if __name__ == '__main__':
arguments1 = zip([222e6, 22e6] * 2)
arguments2 = zip([111e6, 11e6] * 2)
with Pool(4) as pool:
results = pool.starmap_async(busy_foo, arguments1)
results2 = pool.starmap_async(busy_foo, arguments2)
print(results.get())
print(results2.get())
示例输出:
3182 returning: 22000000
3185 returning: 22000000
3185 returning: 11000000
3182 returning: 111000000
3182 returning: 11000000
3185 returning: 111000000
3181 returning: 222000000
3184 returning: 222000000
[222000000, 22000000, 222000000, 22000000]
[111000000, 11000000, 111000000, 11000000]
Process finished with exit code 0
请注意,以较简单的任务结束的进程 3182 和 3185 会立即从第二个参数列表中的任务开始,而无需等待 3181 和 3184 首先完成。
如果出于某种原因,您真的想在每个进程处理一定数量的任务后使用新进程,可以使用 Pool
的 maxtasksperchild
参数。在那里你可以指定在多少任务之后池应该用新的进程替换旧的进程。此参数的默认值为 None
,因此默认情况下 Pool 不会替换进程。
关于python - 多处理。池 : How to start new processes as old ones finish?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/52616895/