Warten höchstens N Sekunden für ein fork'd Kind PID zu beenden

Question

Ich versuche os.fork zu verwenden, os.waitpid und Gewinde einige zeitraubende Arbeit, um ein Kind Prozess zu schicken, wartet höchstens eine bestimmte Menge an Zeit für die Fertigstellung, und dann weitermachen und im Hintergrund laufen lassen. Es gibt auch einen sekundären Timer in dem Kind, das mich vom Laufen zu lange im Hintergrund verhindert, etwa so:

Fork execution 
In the fork's child: 
    Start a thread to execute a task 
     If that thread runs for longer than X milliseconds: 
      thread.stop(), a custom method that softly stops the thread. 
In the fork's parent: 
    If the child's PID stays running for longer than Y milliseconds: 
     return -1 
    else: 
     return 0 

Nach Wert zurückkehrt, mag ich das Skript zu beenden. Das Kind kann und sollte weiterlaufen, wenn es nicht getan wird.

Der Code, den ich versucht habe (abgekürzt) ist:

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() #Sleeps for 30 seconds and ends. 
     thread1.start() 
     print "Started 1!" 
     timeout = X # say, 1000ms 
     while timeout > 0: 
      if not thread1.is_alive(): return "custom thread finished before the deadline!" 
      timeout -= 1 
      sleep(0.001) 
     if thread1.is_alive(): 
      return "custom thread didn't finish before the deadline!" 
      thread1.stop() 
     exit() 
    else: 
     print pid 
     thread2 = Thread(target = os.waitpid, args = (pid, 0)) 
     thread2.start() 
     print "Started 2!" 
     timeout2 = Y # say, 500ms 
     while timeout2 > 0: 
      if not thread2.is_alive(): return "child PID finished!" 
      timeout2 -= 1 
      sleep(0.001) 
     if thread2.is_alive(): 
      return "child PID didn't finish yet!" 
print test() 
print "all done!" 

Der Ausgang richtig ist, dass ich

bekommen

1171 
Started 2! 
Started 1! 
child PID didn't finish yet! 
all done! 
custom thread didn't finish before the deadline! 
all done! 

Aber dann das Skript nicht beendet! Es schläft für die restlichen 28 Sekunden vor

Wie mache ich die Hauptausführung dieses Skript abgeschlossen, nachdem die fork Ed Eltern einen Wert zurückgibt? Wie ich bereits sagte, kann und sollte das Kind im Hintergrund laufen, es sollte die Ausführung der nächsten Aufgabe auf dem Elternteil nicht blockieren.

Ich interessiere mich wirklich nicht, wenn das Kind Ausgabe zum Standardausgang drucken kann - in der tatsächlichen Implementierung ist alles, was es macht, mit einer Datenbank zu sprechen, also muss es nichts interaktiv berichten. Das übergeordnete Element muss jedoch das untergeordnete Element abfertigen, höchstens Y Sekunden warten, bis das untergeordnete Objekt beendet ist, und dann (soweit der Skript aufgerufen wurde) die Ausführung des Skripts beenden, damit die nächste Aktion ausgeführt werden kann. Der andere Timer (X) ist nicht relevant, denke ich; es existiert nur, um das Kind davon abzuhalten, zu lange im Hintergrund zu laufen.

Irgendwelche Ideen? Ich nähere mich wahrscheinlich diesem total falsch, also "neu anfangen und es tun _ Weg" Ideen sind willkommen.

Answer 1

Ich dachte mir, danke für alle Hinweise! Bearbeiten: behoben is_pid_alive Funktion zu arbeiten, wenn auch innerhalb eines Kindes aufgerufen.

Das Problem war, dass der "Watcher" Thread des Elternteils nie abgeschlossen wurde, da os.waitpid keine pollable/loopable Funktion ist. Die Lösung war die "Beobachter" Thread, zu entfernen und stattdessen eine Abfrageschleife zu implementieren, die eine pid_is_alive() Funktion jede Millisekunde überprüft, etwa 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() #Sleeps for 30 seconds and ends. 
     thread1.start() 
     print "Started 1!" 
     timeout = X # say, 1000ms 
     while timeout > 0: 
      if not thread1.is_alive(): return "custom thread finished before the deadline!" 
      timeout -= 1 
      sleep(0.001) 
     if thread1.is_alive(): 
      return "custom thread didn't finish before the deadline!" 
      thread1.stop() 
     exit() 

    else: 
     timeout2 = Y # say, 500ms 
     while timeout2 > 0: 
      if not pid_is_alive(pid): return "child PID finished!" 
      timeout2 -= 1 
      sleep(0.001) 
     if pid_is_alive(pid): 
      print "child PID didn't finish yet!" 
      exit() 
print test() 
print "all done!" 

Answer 2

Dies ist keine exakte Antwort auf Ihre Frage, sondern eine Idee "von Anfang an beginnen und es _ Weg".

Sie könnten das Modul multiprocessing verwenden. Die Funktion Pool.apply_async() ermöglicht es, eine Funktion im Hintergrund auszuführen, und die zurückgegebenen AsyncResult Objektfunktionen wait() und get() Methoden mit einem timeout Parameter.

Ihr Code würde im Wesentlichen wird (ungetestet)

import multiprocessing 

def computationally_intensive(): 
    # whatever 

p = multiprocessing.Pool(1) 
deferred = p.apply_async(computationally_intensive) 
try: 
    result = deferred.get(10)    # 10 being the timeout 
except multiprocessing.TimeoutError: 
    # Handle time out 
# ... 
p.terminate() 

Answer 3

Sie join Methode des Thread

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

Answer 4

Versuchen Sie diesen verwenden können, ist es nicht Einfädeln nicht verwendet, nur reine Gabel/waitpid/alarm/sighandler:

child_exited = False 

def sigh(signum, frame): 
    global child_exited 
    if signum == signal.SIGALRM: 
     print "custom thread didn't finish before the deadline!" 
     #forced exit: 
     exit(1)  

    if signum == signal.SIGCHLD: 
     (pid, status) = os.waitpid(-1, 0) 
     print "child exited with 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) 

     #do work: 
     print "Started 1" 
     time.sleep(60) 

     #clean exit: 
     exit(0) 
    elif (pid > 0): 
     signal.signal(signal.SIGCHLD, sigh) 
     print "Started 2" 
     #this sleep will return prematurely if the child exits 
     time.sleep(10) 

     if not child_exited: 
      print "child PID didn't finish yet!" 
    else: 
     print "fork() failed" 

print test() 
print "all done!"