In Python, how do I know when a process is finishe

Question

From within a Python GUI (PyGTK) I start a process (using multiprocessing). The process takes a long time (~20 minutes) to finish. When the process is finished I would like to clean it up (extract the results and join the process). How do I know when the process has finished?

My colleague suggested a busy loop within the parent process that checks if the child process has finished. Surely there is a better way.

In Unix, when a process is forked, a signal handler is called from within the parent process when the child process has finished. But I cannot see anything like that in Python. Am I missing something?

How is it that the end of a child process can be observed from within the parent process? (Of course, I do not want to call Process.join() as it would freeze up the GUI interface.)

This question is not limited to multi-processing: I have exactly the same problem with multi-threading.

Answer 1

This answer is really simple! (It just took me days to work it out.)

Combined with PyGTK's idle_add(), you can create an AutoJoiningThread. The total code is borderline trivial:

class AutoJoiningThread(threading.Thread):
    def run(self):
        threading.Thread.run(self)
        gobject.idle_add(self.join)

If you want to do more than just join (such as collecting results) then you can extend the above class to emit signals on completion, as is done in the following example:

import threading
import time
import sys
import gobject
gobject.threads_init()

class Child:
    def __init__(self):
        self.result = None

    def play(self, count):
        print "Child starting to play."
        for i in range(count):
            print "Child playing."
            time.sleep(1)
        print "Child finished playing."
        self.result = 42

    def get_result(self, obj):
        print "The result was "+str(self.result)

class AutoJoiningThread(threading.Thread, gobject.GObject):
    __gsignals__ = {
        'finished': (gobject.SIGNAL_RUN_LAST,
                     gobject.TYPE_NONE,
                     ())
        }

    def __init__(self, *args, **kwargs):
        threading.Thread.__init__(self, *args, **kwargs)
        gobject.GObject.__init__(self)

    def run(self):
        threading.Thread.run(self)
        gobject.idle_add(self.join)
        gobject.idle_add(self.emit, 'finished')

    def join(self):
        threading.Thread.join(self)
        print "Called Thread.join()"

if __name__ == '__main__':
    print "Creating child"
    child = Child()
    print "Creating thread"
    thread = AutoJoiningThread(target=child.play,
                               args=(3,))
    thread.connect('finished', child.get_result)
    print "Starting thread"
    thread.start()
    print "Running mainloop (Ctrl+C to exit)"
    mainloop = gobject.MainLoop()

    try:
        mainloop.run()
    except KeyboardInterrupt:
        print "Received KeyboardInterrupt.  Quiting."
        sys.exit()

    print "God knows how we got here.  Quiting."
    sys.exit()

The output of the above example will depend on the order the threads are executed, but it will be similar to:

Creating child
Creating thread
Starting thread
Child starting to play.
 Child playing.
Running mainloop (Ctrl+C to exit)
Child playing.
Child playing.
Child finished playing.
Called Thread.join()
The result was 42
^CReceived KeyboardInterrupt.  Quiting.

It's not possible to create an AutoJoiningProcess in the same way (because we cannot call idle_add() across two different processes), however we can use an AutoJoiningThread to get what we want:

class AutoJoiningProcess(multiprocessing.Process):
    def start(self):
        thread = AutoJoiningThread(target=self.start_process)
        thread.start() # automatically joins

    def start_process(self):
        multiprocessing.Process.start(self)
        self.join()

To demonstrate AutoJoiningProcess here is another example:

import threading
import multiprocessing
import time
import sys
import gobject
gobject.threads_init()

class Child:
    def __init__(self):
        self.result = multiprocessing.Manager().list()

    def play(self, count):
        print "Child starting to play."
        for i in range(count):
            print "Child playing."
            time.sleep(1)
    print "Child finished playing."
        self.result.append(42)

    def get_result(self, obj):
        print "The result was "+str(self.result)

class AutoJoiningThread(threading.Thread, gobject.GObject):
    __gsignals__ = {
        'finished': (gobject.SIGNAL_RUN_LAST,
                     gobject.TYPE_NONE,
                     ())
    }

    def __init__(self, *args, **kwargs):
        threading.Thread.__init__(self, *args, **kwargs)
        gobject.GObject.__init__(self)

    def run(self):
        threading.Thread.run(self)
        gobject.idle_add(self.join)
        gobject.idle_add(self.emit, 'finished')

    def join(self):
        threading.Thread.join(self)
        print "Called Thread.join()"

class AutoJoiningProcess(multiprocessing.Process, gobject.GObject):
    __gsignals__ = {
        'finished': (gobject.SIGNAL_RUN_LAST,
                     gobject.TYPE_NONE,
                     ())
        }

    def __init__(self, *args, **kwargs):
        multiprocessing.Process.__init__(self, *args, **kwargs)
        gobject.GObject.__init__(self)

    def start(self):
        thread = AutoJoiningThread(target=self.start_process)
        thread.start()

    def start_process(self):
        multiprocessing.Process.start(self)
        self.join()
        gobject.idle_add(self.emit, 'finished')

    def join(self):
        multiprocessing.Process.join(self)
        print "Called Process.join()"

if __name__ == '__main__':
    print "Creating child"
    child = Child()
    print "Creating thread"
    process = AutoJoiningProcess(target=child.play,
                               args=(3,))
    process.connect('finished',child.get_result)
    print "Starting thread"
    process.start()
    print "Running mainloop (Ctrl+C to exit)"
    mainloop = gobject.MainLoop()

    try:
        mainloop.run()
    except KeyboardInterrupt:
        print "Received KeyboardInterrupt.  Quiting."
        sys.exit()

    print "God knows how we got here.  Quiting."
    sys.exit()

The resulting output will be very similar to the example above, except this time we have both the process joining and it's attendant thread joining too:

Creating child
Creating thread
Starting thread
Running mainloop (Ctrl+C to exit)
 Child starting to play.
Child playing.
Child playing.
Child playing.
Child finished playing.
Called Process.join()
The result was [42]
Called Thread.join()
^CReceived KeyboardInterrupt.  Quiting.

Unfortunately:

1. This solution is dependent on gobject, due to the use of idle_add(). gobject is used by PyGTK.
2. This is not a true parent/child relationship. If one of these threads is started by another thread, then it will nonetheless be joined by the thread running the mainloop, not the parent thread. This problem holds true for AutoJoiningProcess too, except there I imagine an exception would be thrown.

Thus to use this approach, it would be best to only create threads/process from within the mainloop/GUI.

Answer 2

You can use a queue to communicate with child processes. You can stick intermediate results on it, or messages indicating that milestones have been hit (for progress bars) or just a message indicating that the process is ready to be joined. Polling it with empty is easy and fast.

If you really only want to know if it's done, you can watch the exitcode of your process or poll is_alive().

Answer 3

In my efforts to try to find an answer to my own question, I stumbled across PyGTK's idle_add() function. This gives me the following possibility:

1. Create a new child process that communicates via a Queue.
2. Create a listener thread that listens to the Queue, when the child process sends the listener a message saying that it is finished, the listener calls idle_add() that sets up a callback.
3. During the next time around the main loop the parent process will call the callback.
4. The callback can extract results, join the child process and join the listener-thread.

This seems an overly complex way to re-create Unix's call-callback-when-child-process-is-done.

This must be an uber-common problem with GUIs in Python. Surely there is a standard pattern to solve this problem?

Answer 4

I think as a part of making python multi-platform, simple things like SIGCHLD must be done yourself. Agreed, this is a little more work when all you want to do is know when the child is done, but it really isn't THAT painful. Consider the following that uses a child process to do the work, two multiprocessing.Event instances, and a thread to check if the child process is done:

import threading
from multiprocessing import Process, Event
from time import sleep

def childsPlay(event):
    print "Child started"
    for i in range(3):
        print "Child is playing..."
        sleep(1)
    print "Child done"
    event.set()

def checkChild(event, killEvent):
    event.wait()
    print "Child checked, and is done playing"
    if raw_input("Do again? y/n:") == "y":
        event.clear()
        t = threading.Thread(target=checkChild, args=(event, killEvent))
        t.start()
        p = Process(target=childsPlay, args=(event,))
        p.start()
    else:
        cleanChild()
        killEvent.set()

def cleanChild():
    print "Cleaning up the child..."

if __name__ == '__main__':
    event = Event()
    killEvent = Event()

    # process to do work
    p = Process(target=childsPlay, args=(event,))
    p.start()

    # thread to check on child process
    t = threading.Thread(target=checkChild, args=(event, killEvent))
    t.start()

    try:
        while not killEvent.is_set():
            print "GUI running..."
            sleep(1)
    except KeyboardInterrupt:
        print "Quitting..."
        exit(0)
    finally:
        print "Main done"

EDIT

Joining to all processes and threads created is a good practice because it will help indicate when zombie (never-finishing) processes/threads are being created. I've altered the above code making a ChildChecker class that inherits from threading.Thread. It's sole purpose is to start a job in a separate process, wait for that process to finish, and then notify the GUI when everything is complete. Joining on the ChildChecker will also join the process it is "checking". Now, if the process doesn't join after 5 seconds, the thread will force terminate the process. Enter "y" creates starts a child process running "endlessChildsPlay" that must demonstrate force termination.

import threading
from multiprocessing import Process, Event
from time import sleep

def childsPlay(event):
    print "Child started"
    for i in range(3):
        print "Child is playing..."
        sleep(1)
    print "Child done"
    event.set()

def endlessChildsPlay(event):
    print "Endless child started"
    while True:
        print "Endless child is playing..."
        sleep(1)
        event.set()
    print "Endless child done"

class ChildChecker(threading.Thread):
    def __init__(self, killEvent):
        super(ChildChecker, self).__init__()
        self.killEvent = killEvent
        self.event = Event()
        self.process = Process(target=childsPlay, args=(self.event,))

    def run(self):
        self.process.start()

        while not self.killEvent.is_set():
            self.event.wait()
            print "Child checked, and is done playing"
            if raw_input("Do again? y/n:") == "y":
                self.event.clear()
                self.process = Process(target=endlessChildsPlay, args=(self.event,))
                self.process.start()
            else:
                self.cleanChild()
                self.killEvent.set()

    def join(self):
        print "Joining child process"
        # Timeout on 5 seconds
        self.process.join(5)

        if self.process.is_alive():
            print "Child did not join!  Killing.."
            self.process.terminate()
        print "Joining ChildChecker thread"
        super(ChildChecker, self).join()


    def cleanChild(self):
        print "Cleaning up the child..."

if __name__ == '__main__':
    killEvent = Event()
    # thread to check on child process
    t = ChildChecker(killEvent)
    t.start()

    try:
        while not killEvent.is_set():
            print "GUI running..."
            sleep(1)
    except KeyboardInterrupt:
        print "Quitting..."
        exit(0)
    finally:
        t.join()
        print "Main done"

Answer 5

have a look at the subprocess module:

http://docs.python.org/library/subprocess.html

import subprocess
let pipe = subprocess.Popen("ls -l", stdout=subprocess.PIPE)
allText = pipe.stdout.read()
pipe.wait()
retVal = pipe.returncode