如何从执行中暂停多处理池

Question

使用时:

def myFunction(arg):
    for i in range(10000):
        pass

from multiprocessing import Pool
pool = Pool(processes=3)
pool.map_async( myFunction, ['first','second','third'] )

我希望用户能够在池启动后的任何给定时间暂停执行多处理池.然后我希望用户能够取消暂停(继续)池中的其余项目.怎么实现呢？

编辑:

以下是Blckknght发布的建议的工作实施.谢谢Blckknght!

import multiprocessing
from PyQt4 import QtGui, QtCore


def setup(event):
    global unpaused
    unpaused = event

def myFunction( arg=None):
    unpaused.wait()
    print "Task started...", arg
    for i in range(15000000):
        pass
    print '...task completed.', arg


class MyApp(object):

    def __init__(self):
        super(MyApp, self).__init__()

        app = QtGui.QApplication(sys.argv)
        self.mainWidget = QtGui.QWidget()
        self.mainLayout = QtGui.QVBoxLayout()
        self.mainWidget.setLayout(self.mainLayout)

        self.groupbox = QtGui.QGroupBox()
        self.layout = QtGui.QVBoxLayout()
        self.groupbox.setLayout(self.layout)

        self.pauseButton = QtGui.QPushButton('Pause')
        self.pauseButton.clicked.connect(self.pauseButtonClicked)      
        self.layout.addWidget(self.pauseButton) 

        self.okButton = QtGui.QPushButton('Start Pool')
        self.okButton.clicked.connect(self.startPool) 
        self.layout.addWidget(self.okButton)
        self.layout.addWidget(self.pauseButton)      

        self.mainLayout.addWidget(self.groupbox)
        self.mainWidget.show()
        sys.exit(app.exec_())


    def startPool(self):
        self.event = multiprocessing.Event()
        self.pool=multiprocessing.Pool(1, setup, (self.event,))
        self.result=self.pool.map_async(myFunction, [1,2,3,4,5,6,7,8,9,10])
        self.event.set()
        # self.result.wait()       

    def pauseJob(self):
        self.event.clear()

    def continueJob(self):
        self.event.set()

    def pauseButtonClicked(self):
        if self.pauseButton.text()=='Pause':
            print '\n\t\t ...pausing job...','\n'
            self.pauseButton.setText('Resume')
            self.pauseJob()
        else:
            print '\n\t\t ...resuming job...','\n'
            self.pauseButton.setText('Pause')            
            self.continueJob()

if __name__ == '__main__':
    MyApp()

Answer 1

听起来你想用a multiprocessing.Event来控制你的worker函数的运行.您可以创建一个,然后将其传递给initializer池中的一个,然后等待它myFunction.

这是一个运行工人的例子,每秒都会打印出他们的参数.工作人员可以通过clear事件暂停,然后set再次重新启动.

from time import sleep
import multiprocessing

def setup(event):
    global unpaused
    unpaused = event

def myFunction(arg):
    for i in range(10):
        unpaused.wait()
        print(arg)
        sleep(1)

if __name__ == "__main__":
    event = multiprocessing.Event() # initially unset, so workers will be paused at first
    pool = multiprocessing.Pool(3, setup, (event,))
    result = pool.map_async(myFunction, ["foo", "bar", "baz"])
    event.set()   # unpause workers
    sleep(5)
    event.clear() # pause after five seconds
    sleep(5)
    event.set()   # unpause again after five more seconds
    result.wait() # wait for the rest of the work to be completed

工作进程应打印"foo","bar"和"baz"十次每个,与每次重复之间的一秒的延迟.工作人员将在前五秒钟后暂停,并在五秒钟后重新启动.根据您的实际使用情况,可能有多种方法可以改进此代码,但希望这足以让您朝着正确的方向前进.