python - 如何暂停执行多处理池

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

听起来您想使用 multiprocessing.Event 来控制辅助函数的运行。您可以创建一个，然后将其传递给池的 initializer，然后在 myFunction 中等待它。

这是一个示例，它运行每秒打印一次参数的工作线程。工作人员可以通过清除事件暂停，并通过再次设置重新启动。

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" 各十次，中间有一秒的延迟每次重复。不过，工作人员将在前五秒后暂停，并在再过五秒后重新启动。可能有多种方法可以改进此代码，具体取决于您的实际用例，但希望它足以让您朝着正确的方向前进。