Wie man die Ausführung eines Multitasking-Pools anhält

Question

Beim Verwenden:

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

from multiprocessing import Pool
pool = Pool(processes=3)
pool.map_async( meineFunktion, ['erster','zweiter','dritter'] )

Ich möchte, dass der Benutzer die Ausführung des Pools der Mehrfachverarbeitung zu einem beliebigen Zeitpunkt nach dem Start des Pools anhalten kann. Dann möchte ich, dass der Benutzer (fortsetzen) mit dem Rest der Elemente in einem Pool fortfahren kann. Wie kann man das erreichen?

BEARBEITEN:

Hier ist die funktionierende Implementierung der Vorschläge von Blckknght. Vielen Dank, Blckknght!

import multiprocessing
from PyQt4 import QtGui, QtCore

def setup(event):
    global unpaused
    unpaused = event

def meineFunktion( arg=None):
    unpaused.wait()
    print "Aufgabe gestartet...", arg
    for i in range(15000000):
        pass
    print '...Aufgabe abgeschlossen.', 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(meineFunktion, [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 ...Job pausiert...','\n'
            self.pauseButton.setText('Fortsetzen')
            self.pauseJob()
        else:
            print '\n\t\t ...Job wird fortgesetzt...','\n'
            self.pauseButton.setText('Pause')            
            self.continueJob()

if __name__ == '__main__':
    MyApp()

Answer 1

Es klingt so, als ob du ein multiprocessing.Event verwenden möchtest, um die Ausführung deiner Worker-Funktion zu steuern. Du kannst eine erstellen, dann sie einem initializer des Pools übergeben und anschließend in myFunction darauf warten.

Hier ist ein Beispiel, das Worker ausführt, die alle Sekunde ihr Argument ausgeben. Die Worker können durch clear pausiert und durch erneutes set wieder gestartet werden.

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() # zuerst nicht gesetzt, also werden Worker anfangs pausiert
    pool = multiprocessing.Pool(3, setup, (event,))
    result = pool.map_async(myFunction, ["foo", "bar", "baz"])
    event.set()   # Worker fortsetzen
    sleep(5)
    event.clear() # nach fünf Sekunden pausieren
    sleep(5)
    event.set()   # erneutes Fortsetzen nach weiteren fünf Sekunden
    result.wait() # Warten, bis die restliche Arbeit abgeschlossen ist

Die Worker-Prozesse sollten "foo", "bar" und "baz" jeweils zehnmal ausgeben, mit einer Verzögerung von einer Sekunde zwischen jeder Wiederholung. Die Worker werden jedoch nach den ersten fünf Sekunden pausiert und nach weiteren fünf Sekunden wieder gestartet. Es gibt wahrscheinlich verschiedene Möglichkeiten, diesen Code zu verbessern, abhängig von deinem tatsächlichen Anwendungsfall, aber hoffentlich ist es genug, um dich in die richtige Richtung zu leiten.