Warten Sie maximal N Sekunden darauf, dass ein abgespaltener Kind-PID beendet wird.

Question

Ich versuche, os.fork, os.waitpid und das Threaden zu verwenden, um einige zeitaufwändige Arbeiten an einen Kindprozess zu übergeben, höchstens eine bestimmte Zeit auf dessen Abschluss zu warten und dann fortzufahren, während es im Hintergrund läuft. Im Kind gibt es auch einen sekundären Timer, der verhindert, dass es zu lange im Hintergrund läuft, und zwar so:

Ausführung des Forks
Im Kind des Forks:
    Starten eines Threads, um eine Aufgabe auszuführen
        Wenn dieser Thread länger als X Millisekunden läuft:
             thread.stop(), eine benutzerdefinierte Methode, die den Thread sanft stoppt.
Im Elternteil des Forks:
    Wenn die PID des Kindes länger als Y Millisekunden läuft:
        return -1
    sonst:
        return 0

Nachdem ein Wert zurückgegeben wurde, möchte ich, dass das Skript beendet wird. Das Kind kann und sollte weiterlaufen, wenn es noch nicht beendet ist.

Der von mir ausprobierte Code (verkürzt) lautet:

class MyCustomThread(threading.Thread):
    abort = False
    def run(self):
        counter = 0
        while True:
            if self.abort: break
            counter += 1
            if counter == 30: self.stop()
            sleep(1)
        return 0

    def stop(self):
        self.abort = True

def test():
    X = 2000
    Y = 500
    pid = os.fork()
    if pid == 0:
        thread1 = MyCustomThread() #Schläft 30 Sekunden und endet.
        thread1.start()
        print "Gestartet 1!"
        timeout = X # sagen wir, 1000ms
        while timeout > 0:
            if not thread1.is_alive(): return "benutzerdefinierter Thread wurde vor dem Ablauf der Frist beendet!"
            timeout -= 1
            sleep(0.001)
        if thread1.is_alive():
            return "benutzerdefinierter Thread wurde nicht vor Ablauf der Frist beendet!"
            thread1.stop()
        exit()
    else:
        print pid
        thread2 = Thread(target = os.waitpid, args = (pid, 0))
        thread2.start()
        print "Gestartet 2!"
        timeout2 = Y # sagen wir, 500ms
        while timeout2 > 0:
            if not thread2.is_alive(): return "Kind-PID beendet!"
            timeout2 -= 1
            sleep(0.001)
        if thread2.is_alive():
            return "Kind-PID wurde noch nicht beendet!"
print test()
print "alle erledigt!"

Die Ausgabe ist korrekt, denn ich erhalte

1171
Gestartet 2!
Gestartet 1!
Kind-PID wurde noch nicht beendet!
alle erledigt!
Benutzerdefinierter Thread wurde nicht vor Ablauf der Frist beendet!
alle erledigt!

Aber dann beendet sich das Skript nicht! Es schläft die verbleibenden 28 Sekunden, bevor es

Wie mache ich die Hauptausführung dieses Skripts nachdem der forkte Elternteil einen Wert zurückgibt, beendet? Wie ich bereits sagte, sollte das Kind im Hintergrund weiterlaufen können und auch, aber die Ausführung der nächsten Aufgabe im Elternteil nicht blockieren.

Mir ist es wirklich egal, ob das Kind Ausgaben auf die Standardausgabe drucken kann - in der tatsächlichen Implementierung spricht es nur mit einer Datenbank, also muss es keine interaktiven Meldungen ausgeben. Der Elternteil muss jedoch das Kind beauftragen, höchstens Y Sekunden auf dessen Beendigung zu warten und dann (so weit wie derjenige, der das Skript aufgerufen hat, betroffen ist) die Ausführung des Skripts beenden, damit die nächste Aufgabe erledigt werden kann. Der andere Timer (X) ist meines Erachtens nicht relevant; er dient nur dazu, das Kind daran zu hindern, zu lange im Hintergrund zu laufen.

Irgendwelche Ideen? Wahrscheinlich gehe ich das Ganze total falsch an, also sind Ideen zum "Neuanfang und so machen" willkommen.

Answer 1

Versuchen Sie es mit diesem, es verwendet kein Threading, sondern nur reine Fork/Waitpid/Alarm/Sighandler:

child_exited = False

def sigh(signum, frame):
    global child_exited
    if signum == signal.SIGALRM:
        print "Benutzerdefinierter Thread hat die Frist nicht eingehalten!"
        #Erzwungenes Beenden:
        exit(1)     

    if signum == signal.SIGCHLD:
        (pid, status) = os.waitpid(-1, 0) 
        print "Kind beendet mit Status: " + str(os.WEXITSTATUS(status))
        child_exited = True

def test():
    global child_exited
    pid = os.fork()
    if pid == 0:
        signal.signal(signal.SIGALRM, sigh)
        signal.alarm(30)

        #Arbeit erledigen:
        print "Gestartet 1"
        time.sleep(60)

        #Sauberes Beenden:
        exit(0)
    elif (pid > 0):
        signal.signal(signal.SIGCHLD, sigh)
        print "Gestartet 2"
        #Dieser Sleep wird vorzeitig zurückkehren, wenn das Kind beendet wird
        time.sleep(10)

        if not child_exited:
          print "Kind-PID ist noch nicht fertig!"
    else:
        print "Fehler beim fork()"

print test()
print "Alles erledigt!"

Answer 2

Dies ist keine genaue Antwort auf Ihre Frage, sondern eher eine Idee zum "Neustart und Durchführen auf die _-Weise".

Sie könnten das multiprocessing-Modul verwenden. Die Funktion Pool.apply_async() ermöglicht die Ausführung einer Funktion im Hintergrund, und das zurückgegebene AsyncResult-Objekt verfügt über die Methoden wait() und get() mit einem timeout-Parameter.

Ihr Code würde im Wesentlichen wie folgt aussehen (nicht getestet)

import multiprocessing

def rechenintensiv():
    #was auch immer

p = multiprocessing.Pool(1)
verschoben = p.apply_async(rechenintensiv)
try:
    ergebnis = verschoben.get(10)              # 10 ist der Timeout
except multiprocessing.TimeoutError:
    # Timeout behandeln
# ...
p.terminate()

Answer 3

Ich habe es herausgefunden, danke für die ganzen Hinweise! Bearbeitung: Fehler in der is_pid_alive Funktion behoben, damit sie auch innerhalb eines Kindes funktioniert.

Das Problem war, dass der "Watcher"-Thread des Elternteils nie abgeschlossen wurde, da os.waitpid keine abfragbare / loopbare Funktion ist. Die Lösung bestand darin, den "Watcher"-Thread zu entfernen und stattdessen eine Abfrage-Schleife zu implementieren, die eine pid_is_alive() Funktion jede Millisekunde überprüft, so:

def pid_is_alive(pid):
    try:
        os.kill(pid, 0)
        os.waitpid(pid, os.WNOHANG)
        os.kill(pid, 0)
    except OSError:
        return False
    return True

def test():
    X = 1000 * 1000
    Y = 5000
    pid = os.fork()
    if pid == 0:
        thread1 =  MyCustomThread() #Schläft für 30 Sekunden und endet.
        thread1.start()
        print "Gestartet 1!"
        timeout = X # sagen wir, 1000ms
        while timeout > 0:
            if not thread1.is_alive(): return "benutzerdefinierter Thread vor dem Zeitplan beendet!"
            timeout -= 1
            sleep(0.001)
        if thread1.is_alive():
            return "benutzerdefinierter Thread wurde nicht vor dem Zeitplan beendet!"
            thread1.stop()
        exit()

    else:
        timeout2 = Y # sagen wir, 500ms
        while timeout2 > 0:
            if not pid_is_alive(pid): return "Kind-PID beendet!"
            timeout2 -= 1
            sleep(0.001)
        if pid_is_alive(pid):
            print "Kind-PID ist noch nicht beendet!"
            exit()
print test()
print "alles erledigt!"

Answer 4

Sie können die join Methode des Threads verwenden

thread2.join(timeout=Y)
if not thread2.is_alive():
   # TODO: Nachverarbeitung